Repository: TobikoData/sqlmesh
Pull request #3165 (2024-09-23 19:08:22)

Fix: regression caused by optionally normalizing identifiers in renderer
This PR reverts the `normalize_identifiers` changes introduced in https://github.com/TobikoData/sqlmesh/pull/3077. Switching off identifier normalization has unintended side effects, e.g. `find_tables` treats CTE references like models if there's a casing mismatch, which can lead to more errors.

The motivation was a unit testing issue where we raised an error saying that input data for a model was missing, even though it was a CTE that we knew how to populate.

Not sure if this PR is breaking, please let me know if you think of a case where we'd need a migration and I can update the title accordingly.


Patch:
diff --git a/.circleci/continue_config.yml b/.circleci/continue_config.yml
index 55543e72..42b17a20 100644
--- a/.circleci/continue_config.yml
+++ b/.circleci/continue_config.yml
@@ -331,10 +331,11 @@ workflows:
                 - redshift
                 - bigquery
                 - clickhouse-cloud
+                - athena
           filters:
-            branches:
-              only:
-                - main
+           branches:
+             only:
+               - main
       - trigger_private_tests:
           requires:
             - style_and_slow_tests
diff --git a/.gitignore b/.gitignore
index 64bef2be..f4b62e82 100644
--- a/.gitignore
+++ b/.gitignore
@@ -108,6 +108,7 @@ venv/
 ENV/
 env.bak/
 venv.bak/
+venv*/
 
 # Spyder project settings
 .spyderproject
diff --git a/Makefile b/Makefile
index e0e7390b..a6208060 100644
--- a/Makefile
+++ b/Makefile
@@ -10,7 +10,7 @@ install-doc:
 	pip3 install -r ./docs/requirements.txt
 
 install-engine-test:
-	pip3 install -e ".[dev,web,slack,mysql,postgres,databricks,redshift,bigquery,snowflake,trino,mssql,clickhouse]"
+	pip3 install -e ".[dev,web,slack,mysql,postgres,databricks,redshift,bigquery,snowflake,trino,mssql,clickhouse,athena]"
 
 install-pre-commit:
 	pre-commit install
@@ -209,3 +209,6 @@ redshift-test: guard-REDSHIFT_HOST guard-REDSHIFT_USER guard-REDSHIFT_PASSWORD g
 
 clickhouse-cloud-test: guard-CLICKHOUSE_CLOUD_HOST guard-CLICKHOUSE_CLOUD_USERNAME guard-CLICKHOUSE_CLOUD_PASSWORD engine-clickhouse-install
 	pytest -n auto -x -m "clickhouse_cloud" --retries 3 --junitxml=test-results/junit-clickhouse-cloud.xml
+
+athena-test: guard-AWS_ACCESS_KEY_ID guard-AWS_SECRET_ACCESS_KEY guard-ATHENA_S3_WAREHOUSE_LOCATION engine-athena-install
+	pytest -n auto -x -m "athena" --retries 3 --retry-delay 10 --junitxml=test-results/junit-athena.xml
\ No newline at end of file
diff --git a/docs/guides/configuration.md b/docs/guides/configuration.md
index 9848b9aa..83c697f0 100644
--- a/docs/guides/configuration.md
+++ b/docs/guides/configuration.md
@@ -483,6 +483,7 @@ Example snowflake connection configuration:
 
 These pages describe the connection configuration options for each execution engine.
 
+* [Athena](../integrations/engines/athena.md)
 * [BigQuery](../integrations/engines/bigquery.md)
 * [Databricks](../integrations/engines/databricks.md)
 * [DuckDB](../integrations/engines/duckdb.md)
diff --git a/docs/integrations/engines/athena.md b/docs/integrations/engines/athena.md
new file mode 100644
index 00000000..fbf4fef7
--- /dev/null
+++ b/docs/integrations/engines/athena.md
@@ -0,0 +1,67 @@
+# Athena
+
+## Installation
+
+```
+pip install "sqlmesh[athena]"
+```
+
+## Connection options
+
+### PyAthena connection options
+
+SQLMesh leverages the [PyAthena](https://github.com/laughingman7743/PyAthena) DBAPI driver to connect to Athena. Therefore, the connection options relate to the PyAthena connection options.
+Note that PyAthena uses [boto3](https://boto3.amazonaws.com/v1/documentation/api/latest/index.html) under the hood so you can also use [boto3 environment variables](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html#using-environment-variables) for configuration.
+
+| Option                  | Description                                                                                                                                                  |  Type  | Required |
+|-------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------|:------:|:--------:|
+| `type`                  | Engine type name - must be `athena`                                                                                                                          | string |    Y     |
+| `aws_access_key_id`     | The access key for your AWS user                                                                                                                             | string |    N     |
+| `aws_secret_access_key` | The secret key for your AWS user                                                                                                                             | string |    N     |
+| `role_arn`              | The ARN of a role to assume once authenticated                                                                                                               | string |    N     |
+| `role_session_name`     | The session name to use when assuming `role_arn`                                                                                                             | string |    N     |
+| `region_name`           | The AWS region to use                                                                                                                                        | string |    N     |
+| `work_group`            | The Athena [workgroup](https://docs.aws.amazon.com/athena/latest/ug/workgroups-manage-queries-control-costs.html) to send queries to                         | string |    N     |
+| `s3_staging_dir`        | The S3 location for Athena to write query results. Only required if not using `work_group` OR the configured `work_group` doesnt have a results location set | string |    N     |
+| `schema_name`           | The default schema to place objects in if a schema isnt specified. Defaults to `default`                                                                     | string |    N     |
+| `catalog_name`          | The default catalog to place schemas in. Defaults to `AwsDataCatalog`                                                                                        | string |    N     |
+
+### SQLMesh connection options
+
+These options are specific to SQLMesh itself and are not passed to PyAthena
+
+| Option                  | Description                                                                                                                                                                                      | Type   | Required |
+|-------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|----------|
+| `s3_warehouse_location` | Set the base path in S3 where SQLMesh will instruct Athena to place table data. Only required if you arent specifying the location in the model itself. See [S3 Locations](#s3-locations) below. | string | N        |
+
+## Model properties
+
+The Athena adapter recognises the following model [physical_properties](../../concepts/models/overview.md#physical_properties):
+
+| Name              | Description                                                                                                                                                                               | Type   | Default |
+|-------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|--------|---------|
+| `s3_base_location`| `s3://` base URI of where the snapshot tables for this model should be written. Overrides `s3_warehouse_location` if one is configured.                                                   | string |         |
+| `table_type`      | Sets the [table_type](https://docs.aws.amazon.com/athena/latest/ug/create-table-as.html#ctas-table-properties) Athena uses when creating the table. Valid values are `hive` or `iceberg`. | string | `hive`  |
+
+
+## S3 Locations
+When creating tables, Athena needs to know where in S3 the table data is located. You cannot issue a `CREATE TABLE` statement without specifying a `LOCATION` for the table data.
+
+In addition, unlike other engines such as Trino, Athena will not infer a table location if you set a _schema_ location via `CREATE SCHEMA <schema> LOCATION 's3://schema/location'`.
+
+Therefore, in order for SQLMesh to issue correct `CREATE TABLE` statements to Athena, you need to configure where the tables should be stored. There are two options for this:
+
+- **Project-wide:** set `s3_warehouse_location` in the connection config. SQLMesh will set the table `LOCATION` to be `<s3_warehouse_location>/<schema_name>/<snapshot_table_name>` when it creates a snapshot of your model.
+- **Per-model:** set `s3_base_location` in the model `physical_properties`. SQLMesh will set the table `LOCATION` to be `<s3_base_location>/<snapshot_table_name>` every time it creates a snapshot of your model. This takes precedence over any `s3_warehouse_location` set in the connection config.
+
+
+## Limitations
+Athena was initially designed to read data stored in S3 and to do so without changing that data. This means that it does not have good support for mutating tables. In particular, it will not delete data from Hive tables.
+
+Consequently, any SQLMesh model types that needs to delete or merge data from existing tables will not work on Hive tables. In addition, [forward only changes](../../concepts/plans.md#forward-only-change) that mutate the schemas of existing tables have a high chance of failure because Athena supports very limited schema modifications on Hive tables.
+
+However, Athena does support [Apache Iceberg](https://docs.aws.amazon.com/athena/latest/ug/querying-iceberg.html) tables which allow a full range of operations. These can be used for more complex model types such as [`INCREMENTAL_BY_UNIQUE_KEY`](../../concepts/models/model_kinds.md#incremental_by_unique_key) and [`SCD_TYPE_2`](../../concepts/models/model_kinds.md#scd-type-2).
+
+To use an Iceberg table for a model, set `table_type='iceberg'` in the model [physical_properties](../../concepts/models/overview.md#physical_properties).
+
+In general, Iceberg tables offer the most flexibility and you'll run into the least SQLMesh limitations when using them. However, we create Hive tables by default because Athena creates Hive tables by default, so Iceberg tables are opt-in rather than opt-out.
\ No newline at end of file
diff --git a/docs/integrations/engines/gcp-postgres.md b/docs/integrations/engines/gcp-postgres.md
index a63c01ad..e53140e0 100644
--- a/docs/integrations/engines/gcp-postgres.md
+++ b/docs/integrations/engines/gcp-postgres.md
@@ -17,4 +17,6 @@ pip install "sqlmesh[gcppostgres]"
 | `user`                       | The username (posgres or IAM) to use for authentication                             | string  |    Y     |
 | `password`                   | The password to use for authentication. Required when connecting as a Postgres user | string  |    N     |
 | `enable_iam_auth`            | Enables IAM authentication. Required when connecting as an IAM user                 | boolean |    N     |
+| `keyfile`                    | Path to the keyfile to be used with enable_iam_auth instead of ADC                  | string  |    N     |
+| `keyfile_json`               | Keyfile information provided inline (not recommended)                               |  dict   |    N     |
 | `db`                         | The name of the database instance to connect to                                     | string  |    Y     |
diff --git a/docs/integrations/overview.md b/docs/integrations/overview.md
index 7200fc6c..f229137d 100644
--- a/docs/integrations/overview.md
+++ b/docs/integrations/overview.md
@@ -11,6 +11,7 @@ SQLMesh supports integrations with the following tools:
 ## Execution engines
 SQLMesh supports the following execution engines for running SQLMesh projects:
 
+* [Athena](./engines/athena.md)
 * [BigQuery](./engines/bigquery.md)
 * [Databricks](./engines/databricks.md)
 * [DuckDB](./engines/duckdb.md)
diff --git a/mkdocs.yml b/mkdocs.yml
index 7cf1d64f..9e032a10 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -66,6 +66,7 @@ nav:
       - integrations/dbt.md
       - integrations/github.md
     - Execution engines:
+      - integrations/engines/athena.md
       - integrations/engines/bigquery.md
       - integrations/engines/clickhouse.md
       - integrations/engines/databricks.md
diff --git a/setup.py b/setup.py
index 7dfa80ff..f3fb54f3 100644
--- a/setup.py
+++ b/setup.py
@@ -50,6 +50,7 @@ setup(
         "sqlglot[rs]~=25.22.0",
     ],
     extras_require={
+        "athena": ["PyAthena[Pandas]"],
         "bigquery": [
             "google-cloud-bigquery[pandas]",
             "google-cloud-bigquery-storage",
diff --git a/sqlmesh/core/config/common.py b/sqlmesh/core/config/common.py
index 819740c4..ab07133d 100644
--- a/sqlmesh/core/config/common.py
+++ b/sqlmesh/core/config/common.py
@@ -4,7 +4,7 @@ import typing as t
 from enum import Enum
 
 from sqlmesh.utils import classproperty
-from sqlmesh.utils.errors import ConfigError
+from sqlmesh.utils.errors import ConfigError, SQLMeshError
 from sqlmesh.utils.pydantic import field_validator
 
 
@@ -86,3 +86,17 @@ variables_validator = field_validator(
     mode="before",
     check_fields=False,
 )(_variables_validator)
+
+
+def validate_s3_location(value: str, error_type: t.Type[Exception] = SQLMeshError) -> str:
+    if not value.startswith("s3://"):
+        raise error_type(f"Location '{value}' must be a s3:// URI")
+
+    if not value.endswith("/"):
+        value += "/"
+
+    # To avoid HIVE_METASTORE_ERROR: S3 resource path length must be less than or equal to 700.
+    if len(value) > 700:
+        raise error_type(f"Location '{value}' cannot be more than 700 characters")
+
+    return value
diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py
index bc6b10b9..0c790d4a 100644
--- a/sqlmesh/core/config/connection.py
+++ b/sqlmesh/core/config/connection.py
@@ -18,6 +18,7 @@ from sqlmesh.core.config.base import BaseConfig
 from sqlmesh.core.config.common import (
     concurrent_tasks_validator,
     http_headers_validator,
+    validate_s3_location,
 )
 from sqlmesh.core.engine_adapter import EngineAdapter
 from sqlmesh.utils.errors import ConfigError
@@ -848,6 +849,8 @@ class GCPPostgresConnectionConfig(ConnectionConfig):
         password: The postgres user's password. Only needed when the user is a postgres user.
         enable_iam_auth: Set to True when user is an IAM user.
         db: Name of the db to connect to.
+        keyfile: string path to json service account credentials file
+        keyfile_json: dict service account credentials info
         pre_ping: Whether or not to pre-ping the connection before starting a new transaction to ensure it is still alive.
     """
 
@@ -856,8 +859,11 @@ class GCPPostgresConnectionConfig(ConnectionConfig):
     password: t.Optional[str] = None
     enable_iam_auth: t.Optional[bool] = None
     db: str
+    # Keyfile Auth
+    keyfile: t.Optional[str] = None
+    keyfile_json: t.Optional[t.Dict[str, t.Any]] = None
     timeout: t.Optional[int] = None
-
+    scopes: t.Tuple[str, ...] = ("https://www.googleapis.com/auth/sqlservice.admin",)
     driver: str = "pg8000"
     type_: Literal["gcp_postgres"] = Field(alias="type", default="gcp_postgres")
     concurrent_tasks: int = 4
@@ -904,8 +910,19 @@ class GCPPostgresConnectionConfig(ConnectionConfig):
     @property
     def _connection_factory(self) -> t.Callable:
         from google.cloud.sql.connector import Connector
+        from google.oauth2 import service_account
+
+        creds = None
+        if self.keyfile:
+            creds = service_account.Credentials.from_service_account_file(
+                self.keyfile, scopes=self.scopes
+            )
+        elif self.keyfile_json:
+            creds = service_account.Credentials.from_service_account_info(
+                self.keyfile_json, scopes=self.scopes
+            )
 
-        return Connector().connect
+        return Connector(credentials=creds).connect
 
 
 class RedshiftConnectionConfig(ConnectionConfig):
@@ -1467,6 +1484,77 @@ class ClickhouseConnectionConfig(ConnectionConfig):
         return {"compress": compress, "client_name": f"SQLMesh/{__version__}", **settings}
 
 
+class AthenaConnectionConfig(ConnectionConfig):
+    # PyAthena connection options
+    aws_access_key_id: t.Optional[str] = None
+    aws_secret_access_key: t.Optional[str] = None
+    role_arn: t.Optional[str] = None
+    role_session_name: t.Optional[str] = None
+    region_name: t.Optional[str] = None
+    work_group: t.Optional[str] = None
+    s3_staging_dir: t.Optional[str] = None
+    schema_name: t.Optional[str] = None
+    catalog_name: t.Optional[str] = None
+
+    # SQLMesh options
+    s3_warehouse_location: t.Optional[str] = None
+    concurrent_tasks: int = 4
+    register_comments: Literal[False] = (
+        False  # because Athena doesnt support comments in most cases
+    )
+    pre_ping: Literal[False] = False
+
+    type_: Literal["athena"] = Field(alias="type", default="athena")
+
+    @model_validator(mode="after")
+    @model_validator_v1_args
+    def _root_validator(cls, values: t.Dict[str, t.Any]) -> t.Dict[str, t.Any]:
+        work_group = values.get("work_group")
+        s3_staging_dir = values.get("s3_staging_dir")
+        s3_warehouse_location = values.get("s3_warehouse_location")
+
+        if not work_group and not s3_staging_dir:
+            raise ConfigError("At least one of work_group or s3_staging_dir must be set")
+
+        if s3_staging_dir:
+            values["s3_staging_dir"] = validate_s3_location(s3_staging_dir, error_type=ConfigError)
+
+        if s3_warehouse_location:
+            values["s3_warehouse_location"] = validate_s3_location(
+                s3_warehouse_location, error_type=ConfigError
+            )
+
+        return values
+
+    @property
+    def _connection_kwargs_keys(self) -> t.Set[str]:
+        return {
+            "aws_access_key_id",
+            "aws_secret_access_key",
+            "role_arn",
+            "role_session_name",
+            "region_name",
+            "work_group",
+            "s3_staging_dir",
+            "schema_name",
+            "catalog_name",
+        }
+
+    @property
+    def _engine_adapter(self) -> t.Type[EngineAdapter]:
+        return engine_adapter.AthenaEngineAdapter
+
+    @property
+    def _extra_engine_config(self) -> t.Dict[str, t.Any]:
+        return {"s3_warehouse_location": self.s3_warehouse_location}
+
+    @property
+    def _connection_factory(self) -> t.Callable:
+        from pyathena import connect  # type: ignore
+
+        return connect
+
+
 CONNECTION_CONFIG_TO_TYPE = {
     # Map all subclasses of ConnectionConfig to the value of their `type_` field.
     tpe.all_field_infos()["type_"].default: tpe
diff --git a/sqlmesh/core/engine_adapter/__init__.py b/sqlmesh/core/engine_adapter/__init__.py
index 1d2fe878..25c45d2e 100644
--- a/sqlmesh/core/engine_adapter/__init__.py
+++ b/sqlmesh/core/engine_adapter/__init__.py
@@ -17,6 +17,7 @@ from sqlmesh.core.engine_adapter.redshift import RedshiftEngineAdapter
 from sqlmesh.core.engine_adapter.snowflake import SnowflakeEngineAdapter
 from sqlmesh.core.engine_adapter.spark import SparkEngineAdapter
 from sqlmesh.core.engine_adapter.trino import TrinoEngineAdapter
+from sqlmesh.core.engine_adapter.athena import AthenaEngineAdapter
 
 DIALECT_TO_ENGINE_ADAPTER = {
     "hive": SparkEngineAdapter,
@@ -31,6 +32,7 @@ DIALECT_TO_ENGINE_ADAPTER = {
     "mysql": MySQLEngineAdapter,
     "mssql": MSSQLEngineAdapter,
     "trino": TrinoEngineAdapter,
+    "athena": AthenaEngineAdapter,
 }
 
 DIALECT_ALIASES = {
diff --git a/sqlmesh/core/engine_adapter/athena.py b/sqlmesh/core/engine_adapter/athena.py
new file mode 100644
index 00000000..0f23cf1b
--- /dev/null
+++ b/sqlmesh/core/engine_adapter/athena.py
@@ -0,0 +1,403 @@
+from __future__ import annotations
+from functools import lru_cache
+import typing as t
+import logging
+from sqlglot import exp
+from sqlmesh.core.dialect import to_schema
+from sqlmesh.core.config.common import validate_s3_location
+from sqlmesh.core.engine_adapter.mixins import PandasNativeFetchDFSupportMixin
+from sqlmesh.core.engine_adapter.trino import TrinoEngineAdapter
+from sqlmesh.core.node import IntervalUnit
+import os
+from sqlmesh.utils.errors import SQLMeshError
+from sqlmesh.core.engine_adapter.shared import (
+    CatalogSupport,
+    DataObject,
+    DataObjectType,
+    CommentCreationTable,
+    CommentCreationView,
+)
+
+if t.TYPE_CHECKING:
+    from sqlmesh.core._typing import SchemaName, TableName
+
+logger = logging.getLogger(__name__)
+
+
+class AthenaEngineAdapter(PandasNativeFetchDFSupportMixin):
+    DIALECT = "athena"
+    SUPPORTS_TRANSACTIONS = False
+    SUPPORTS_REPLACE_TABLE = False
+    # Athena has the concept of catalogs but no notion of current_catalog or setting the current catalog
+    CATALOG_SUPPORT = CatalogSupport.UNSUPPORTED
+    # Athena's support for table and column comments is too patchy to consider "supported"
+    # Hive tables: Table + Column comments are supported
+    # Iceberg tables: Column comments only
+    # CTAS, Views: No comment support at all
+    COMMENT_CREATION_TABLE = CommentCreationTable.UNSUPPORTED
+    COMMENT_CREATION_VIEW = CommentCreationView.UNSUPPORTED
+    SCHEMA_DIFFER = TrinoEngineAdapter.SCHEMA_DIFFER
+
+    def __init__(
+        self, *args: t.Any, s3_warehouse_location: t.Optional[str] = None, **kwargs: t.Any
+    ):
+        # Need to pass s3_warehouse_location to the superclass so that it goes into _extra_config
+        # which means that EngineAdapter.with_log_level() keeps this property when it makes a clone
+        super().__init__(*args, s3_warehouse_location=s3_warehouse_location, **kwargs)
+        self.s3_warehouse_location = s3_warehouse_location
+
+    @property
+    def s3_warehouse_location(self) -> t.Optional[str]:
+        return self._s3_warehouse_location
+
+    @s3_warehouse_location.setter
+    def s3_warehouse_location(self, value: t.Optional[str]) -> None:
+        if value:
+            value = validate_s3_location(value)
+        self._s3_warehouse_location = value
+
+    def create_state_table(
+        self,
+        table_name: str,
+        columns_to_types: t.Dict[str, exp.DataType],
+        primary_key: t.Optional[t.Tuple[str, ...]] = None,
+    ) -> None:
+        self.create_table(
+            table_name,
+            columns_to_types,
+            primary_key=primary_key,
+            table_properties={
+                # it's painfully slow, but it works
+                "table_type": exp.Literal.string("iceberg")
+            },
+        )
+
+    def _get_data_objects(
+        self, schema_name: SchemaName, object_names: t.Optional[t.Set[str]] = None
+    ) -> t.List[DataObject]:
+        """
+        Returns all the data objects that exist in the given schema and optionally catalog.
+        """
+        schema_name = to_schema(schema_name)
+        schema = schema_name.db
+        query = (
+            exp.select(
+                exp.case()
+                .when(
+                    # calling code expects data objects in the default catalog to have their catalog set to None
+                    exp.column("table_catalog", table="t").eq("awsdatacatalog"),
+                    exp.Null(),
+                )
+                .else_(exp.column("table_catalog"))
+                .as_("catalog"),
+                exp.column("table_schema", table="t").as_("schema"),
+                exp.column("table_name", table="t").as_("name"),
+                exp.case()
+                .when(
+                    exp.column("table_type", table="t").eq("BASE TABLE"),
+                    exp.Literal.string("table"),
+                )
+                .else_(exp.column("table_type", table="t"))
+                .as_("type"),
+            )
+            .from_(exp.to_table("information_schema.tables", alias="t"))
+            .where(exp.column("table_schema", table="t").eq(schema))
+        )
+        if object_names:
+            query = query.where(exp.column("table_name", table="t").isin(*object_names))
+
+        df = self.fetchdf(query)
+
+        return [
+            DataObject(
+                catalog=row.catalog,  # type: ignore
+                schema=row.schema,  # type: ignore
+                name=row.name,  # type: ignore
+                type=DataObjectType.from_str(row.type),  # type: ignore
+            )
+            for row in df.itertuples()
+        ]
+
+    def columns(
+        self, table_name: TableName, include_pseudo_columns: bool = False
+    ) -> t.Dict[str, exp.DataType]:
+        table = exp.to_table(table_name)
+        query = (
+            exp.select("column_name", "data_type")
+            .from_("information_schema.columns")
+            .where(exp.column("table_schema").eq(table.db), exp.column("table_name").eq(table.name))
+            .order_by("ordinal_position")
+        )
+        result = self.fetchdf(query, quote_identifiers=True)
+        return {
+            str(r.column_name): exp.DataType.build(str(r.data_type))
+            for r in result.itertuples(index=False)
+        }
+
+    def _create_schema(
+        self,
+        schema_name: SchemaName,
+        ignore_if_exists: bool,
+        warn_on_error: bool,
+        properties: t.List[exp.Expression],
+        kind: str,
+    ) -> None:
+        if location := self._table_location(table_properties=None, table=exp.to_table(schema_name)):
+            # don't add extra LocationProperty's if one already exists
+            if not any(p for p in properties if isinstance(p, exp.LocationProperty)):
+                properties.append(location)
+
+        return super()._create_schema(
+            schema_name=schema_name,
+            ignore_if_exists=ignore_if_exists,
+            warn_on_error=warn_on_error,
+            properties=properties,
+            kind=kind,
+        )
+
+    def _build_create_table_exp(
+        self,
+        table_name_or_schema: t.Union[exp.Schema, TableName],
+        expression: t.Optional[exp.Expression],
+        exists: bool = True,
+        replace: bool = False,
+        columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+        table_description: t.Optional[str] = None,
+        table_kind: t.Optional[str] = None,
+        partitioned_by: t.Optional[t.List[exp.Expression]] = None,
+        table_properties: t.Optional[t.Dict[str, exp.Expression]] = None,
+        **kwargs: t.Any,
+    ) -> exp.Create:
+        exists = False if replace else exists
+
+        table: exp.Table
+        if isinstance(table_name_or_schema, str):
+            table = exp.to_table(table_name_or_schema)
+        elif isinstance(table_name_or_schema, exp.Schema):
+            table = table_name_or_schema.this
+        else:
+            table = table_name_or_schema
+
+        properties = self._build_table_properties_exp(
+            table=table,
+            expression=expression,
+            columns_to_types=columns_to_types,
+            partitioned_by=partitioned_by,
+            table_properties=table_properties,
+            table_description=table_description,
+            table_kind=table_kind,
+            **kwargs,
+        )
+
+        is_hive = self._table_type(table_properties) == "hive"
+
+        # Filter any PARTITIONED BY properties from the main column list since they cant be specified in both places
+        # ref: https://docs.aws.amazon.com/athena/latest/ug/partitions.html
+        if is_hive and partitioned_by and isinstance(table_name_or_schema, exp.Schema):
+            partitioned_by_column_names = {e.name for e in partitioned_by}
+            filtered_expressions = [
+                e
+                for e in table_name_or_schema.expressions
+                if isinstance(e, exp.ColumnDef) and e.this.name not in partitioned_by_column_names
+            ]
+            table_name_or_schema.args["expressions"] = filtered_expressions
+
+        return exp.Create(
+            this=table_name_or_schema,
+            kind=table_kind or "TABLE",
+            replace=replace,
+            exists=exists,
+            expression=expression,
+            properties=properties,
+        )
+
+    def _build_table_properties_exp(
+        self,
+        catalog_name: t.Optional[str] = None,
+        storage_format: t.Optional[str] = None,
+        partitioned_by: t.Optional[t.List[exp.Expression]] = None,
+        partition_interval_unit: t.Optional[IntervalUnit] = None,
+        clustered_by: t.Optional[t.List[str]] = None,
+        table_properties: t.Optional[t.Dict[str, exp.Expression]] = None,
+        columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+        table_description: t.Optional[str] = None,
+        table_kind: t.Optional[str] = None,
+        table: t.Optional[exp.Table] = None,
+        expression: t.Optional[exp.Expression] = None,
+        **kwargs: t.Any,
+    ) -> t.Optional[exp.Properties]:
+        properties: t.List[exp.Expression] = []
+        table_properties = table_properties or {}
+
+        is_hive = self._table_type(table_properties) == "hive"
+        is_iceberg = not is_hive
+
+        if is_hive and not expression:
+            # Hive tables are CREATE EXTERNAL TABLE, Iceberg tables are CREATE TABLE
+            # Unless it's a CTAS, those are always CREATE TABLE
+            properties.append(exp.ExternalProperty())
+
+        if table_description:
+            properties.append(exp.SchemaCommentProperty(this=exp.Literal.string(table_description)))
+
+        if partitioned_by:
+            schema_expressions: t.List[exp.Expression] = []
+            if is_hive and columns_to_types:
+                # For Hive-style tables, you cannot include the partitioned by columns in the main set of columns
+                # In the PARTITIONED BY expression, you also cant just include the column names, you need to include the data type as well
+                # ref: https://docs.aws.amazon.com/athena/latest/ug/partitions.html
+                for match_name, match_dtype in self._find_matching_columns(
+                    partitioned_by, columns_to_types
+                ):
+                    column_def = exp.ColumnDef(this=exp.to_identifier(match_name), kind=match_dtype)
+                    schema_expressions.append(column_def)
+            else:
+                schema_expressions = partitioned_by
+
+            properties.append(
+                exp.PartitionedByProperty(this=exp.Schema(expressions=schema_expressions))
+            )
+
+        if clustered_by:
+            # Athena itself supports CLUSTERED BY, via the syntax CLUSTERED BY (col) INTO <n> BUCKETS
+            # However, SQLMesh is more closely aligned with BigQuery's notion of clustering and
+            # defines `clustered_by` as a List[str] with no way of indicating the number of buckets
+            #
+            # Athena's concept of CLUSTER BY is more like Iceberg's `bucket(<num_buckets>, col)` partition transform
+            logging.warning("clustered_by is not supported in the Athena adapter at this time")
+
+        if storage_format:
+            if is_iceberg:
+                # TBLPROPERTIES('format'='parquet')
+                table_properties["format"] = exp.Literal.string(storage_format)
+            else:
+                # STORED AS PARQUET
+                properties.append(exp.FileFormatProperty(this=storage_format))
+
+        if table and (location := self._table_location_or_raise(table_properties, table)):
+            properties.append(location)
+
+            if is_iceberg and expression:
+                # To make a CTAS expression persist as iceberg, alongside setting `table_type=iceberg`, you also need to set is_external=false
+                # Note that SQLGlot does the right thing with LocationProperty and writes it as `location` (Iceberg) instead of `external_location` (Hive)
+                # ref: https://docs.aws.amazon.com/athena/latest/ug/create-table-as.html#ctas-table-properties
+                properties.append(exp.Property(this=exp.var("is_external"), value="false"))
+
+        for name, value in table_properties.items():
+            properties.append(exp.Property(this=exp.var(name), value=value))
+
+        if properties:
+            return exp.Properties(expressions=properties)
+
+        return None
+
+    def _truncate_table(self, table_name: TableName) -> None:
+        table = exp.to_table(table_name)
+        # Athena doesnt support TRUNCATE TABLE. The closest thing is DELETE FROM <table> but it only works on Iceberg
+        self.execute(f"DELETE FROM {table.sql(dialect=self.dialect, identify=True)}")
+
+    def _table_type(
+        self, table_properties: t.Optional[t.Dict[str, exp.Expression]] = None
+    ) -> t.Union[t.Literal["hive"], t.Literal["iceberg"]]:
+        """
+        Use the user-specified table_properties to figure out of this is a Hive or an Iceberg table
+        """
+        # if we cant detect any indication of Iceberg, this is a Hive table
+        if table_properties and (table_type := table_properties.get("table_type", None)):
+            if "iceberg" in table_type.sql(dialect=self.dialect).lower():
+                return "iceberg"
+        return "hive"
+
+    @lru_cache()
+    def _query_table_type(
+        self, table_name: TableName
+    ) -> t.Union[t.Literal["hive"], t.Literal["iceberg"]]:
+        """
+        Hit the DB to check if this is a Hive or an Iceberg table
+        """
+        table_name = exp.to_table(table_name)
+        # Note: SHOW TBLPROPERTIES gets parsed by SQLGlot as an exp.Command anyway so we just use a string here
+        # This also means we need to use dialect="hive" instead of dialect="athena" so that the identifiers get the correct quoting (backticks)
+        for row in self.fetchall(f"SHOW TBLPROPERTIES {table_name.sql(dialect='hive')}"):
+            # This query returns a single column with values like 'EXTERNAL\tTRUE'
+            row_lower = row[0].lower()
+            if "external" in row_lower and "true" in row_lower:
+                return "hive"
+        return "iceberg"
+
+    def _table_location_or_raise(
+        self, table_properties: t.Optional[t.Dict[str, exp.Expression]], table: exp.Table
+    ) -> exp.LocationProperty:
+        location = self._table_location(table_properties, table)
+        if not location:
+            raise SQLMeshError(
+                f"Cannot figure out location for table {table}. Please either set `s3_base_location` in `physical_properties` or set `s3_warehouse_location` in the Athena connection config"
+            )
+        return location
+
+    def _table_location(
+        self,
+        table_properties: t.Optional[t.Dict[str, exp.Expression]],
+        table: exp.Table,
+    ) -> t.Optional[exp.LocationProperty]:
+        base_uri: str
+
+        # If the user has manually specified a `s3_base_location`, use it
+        if table_properties and "s3_base_location" in table_properties:
+            s3_base_location_property = table_properties.pop(
+                "s3_base_location"
+            )  # pop because it's handled differently and we dont want it to end up in the TBLPROPERTIES clause
+            if isinstance(s3_base_location_property, exp.Expression):
+                base_uri = s3_base_location_property.name
+            else:
+                base_uri = s3_base_location_property
+
+        elif self.s3_warehouse_location:
+            # If the user has set `s3_warehouse_location` in the connection config, the base URI is <s3_warehouse_location>/<catalog>/<schema>/
+            base_uri = os.path.join(self.s3_warehouse_location, table.catalog or "", table.db or "")
+        else:
+            return None
+
+        full_uri = validate_s3_location(os.path.join(base_uri, table.text("this") or ""))
+        return exp.LocationProperty(this=exp.Literal.string(full_uri))
+
+    def _find_matching_columns(
+        self, partitioned_by: t.List[exp.Expression], columns_to_types: t.Dict[str, exp.DataType]
+    ) -> t.List[t.Tuple[str, exp.DataType]]:
+        matches = []
+        for col in partitioned_by:
+            # TODO: do we care about normalization?
+            key = col.name
+            if isinstance(col, exp.Column) and (match_dtype := columns_to_types.get(key)):
+                matches.append((key, match_dtype))
+        return matches
+
+    def delete_from(self, table_name: TableName, where: t.Union[str, exp.Expression]) -> None:
+        table_type = self._query_table_type(table_name)
+
+        # If Iceberg, DELETE operations work as expected
+        if table_type == "iceberg":
+            return super().delete_from(table_name, where)
+
+        # If Hive, DELETE is an error
+        if table_type == "hive":
+            # However, if the table is empty, we can make DELETE a no-op
+            # This simplifies a bunch of calling code that just assumes DELETE works (which to be fair is a reasonable assumption since it does for every other engine)
+            empty_check = (
+                exp.select("*").from_(table_name).limit(1)
+            )  # deliberately not count(*) because we want the engine to stop as soon as it finds a record
+            if len(self.fetchall(empty_check)) > 0:
+                # TODO: in future, if SQLMesh adds support for explicit partition management, we may
+                # be able to covert the DELETE query into an ALTER TABLE DROP PARTITION assuming the WHERE clause fully covers the partition bounds
+                raise SQLMeshError("Cannot delete from non-empty Hive table")
+
+        return None
+
+    def _drop_object(
+        self,
+        name: TableName | SchemaName,
+        exists: bool = True,
+        kind: str = "TABLE",
+        **drop_args: t.Any,
+    ) -> None:
+        return super()._drop_object(name, exists=exists, kind=kind, **drop_args)
diff --git a/sqlmesh/core/engine_adapter/base.py b/sqlmesh/core/engine_adapter/base.py
index 12d7cabb..cf2e008a 100644
--- a/sqlmesh/core/engine_adapter/base.py
+++ b/sqlmesh/core/engine_adapter/base.py
@@ -1034,8 +1034,9 @@ class EngineAdapter:
         schema_name: SchemaName,
         ignore_if_exists: bool = True,
         warn_on_error: bool = True,
-        properties: t.List[exp.Expression] = [],
+        properties: t.Optional[t.List[exp.Expression]] = None,
     ) -> None:
+        properties = properties or []
         return self._create_schema(
             schema_name=schema_name,
             ignore_if_exists=ignore_if_exists,
@@ -1325,6 +1326,7 @@ class EngineAdapter:
             table_description=table_description,
             column_descriptions=column_descriptions,
             truncate=truncate,
+            **kwargs,
         )
 
     def scd_type_2_by_column(
@@ -1358,6 +1360,7 @@ class EngineAdapter:
             table_description=table_description,
             column_descriptions=column_descriptions,
             truncate=truncate,
+            **kwargs,
         )
 
     def _scd_type_2(
@@ -1377,6 +1380,7 @@ class EngineAdapter:
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         truncate: bool = False,
+        **kwargs: t.Any,
     ) -> None:
         def remove_managed_columns(
             cols_to_types: t.Dict[str, exp.DataType],
@@ -1730,6 +1734,7 @@ class EngineAdapter:
                 columns_to_types=columns_to_types,
                 table_description=table_description,
                 column_descriptions=column_descriptions,
+                **kwargs,
             )
 
     def merge(
diff --git a/sqlmesh/core/engine_adapter/trino.py b/sqlmesh/core/engine_adapter/trino.py
index 82246528..61fe20c3 100644
--- a/sqlmesh/core/engine_adapter/trino.py
+++ b/sqlmesh/core/engine_adapter/trino.py
@@ -223,6 +223,7 @@ class TrinoEngineAdapter(
         table_description: t.Optional[str] = None,
         column_descriptions: t.Optional[t.Dict[str, str]] = None,
         truncate: bool = False,
+        **kwargs: t.Any,
     ) -> None:
         if columns_to_types and self.current_catalog_type == "delta_lake":
             columns_to_types = self._to_delta_ts(columns_to_types)
@@ -243,6 +244,7 @@ class TrinoEngineAdapter(
             table_description,
             column_descriptions,
             truncate,
+            **kwargs,
         )
 
     # delta_lake only supports two timestamp data types. This method converts other
diff --git a/sqlmesh/core/model/definition.py b/sqlmesh/core/model/definition.py
index 01d0c904..88306c63 100644
--- a/sqlmesh/core/model/definition.py
+++ b/sqlmesh/core/model/definition.py
@@ -513,26 +513,11 @@ class _Model(ModelMeta, frozen=True):
         Return:
             The mocked out ctas query.
         """
-        query = self.render_query_or_raise(**render_kwarg).copy()
+        query = self.render_query_or_raise(**render_kwarg).limit(0)
 
         for select_or_set_op in query.find_all(exp.Select, exp.SetOperation):
-            skip_limit = False
-            ancestor = select_or_set_op.parent
-            while ancestor and not skip_limit:
-                if isinstance(ancestor, exp.With) and ancestor.recursive:
-                    skip_limit = True
-                ancestor = ancestor.parent
-
             if isinstance(select_or_set_op, exp.Select) and select_or_set_op.args.get("from"):
                 select_or_set_op.where(exp.false(), copy=False)
-                if not skip_limit and not isinstance(select_or_set_op.parent, exp.SetOperation):
-                    select_or_set_op.limit(0, copy=False)
-            elif (
-                not skip_limit
-                and isinstance(select_or_set_op, exp.SetOperation)
-                and not isinstance(select_or_set_op.parent, exp.SetOperation)
-            ):
-                select_or_set_op.set("limit", exp.Limit(expression=exp.Literal.number(0)))
 
         if self.managed_columns:
             query.select(
diff --git a/sqlmesh/core/renderer.py b/sqlmesh/core/renderer.py
index e395c1f6..c466ed62 100644
--- a/sqlmesh/core/renderer.py
+++ b/sqlmesh/core/renderer.py
@@ -77,7 +77,6 @@ class BaseExpressionRenderer:
         table_mapping: t.Optional[t.Dict[str, str]] = None,
         deployability_index: t.Optional[DeployabilityIndex] = None,
         runtime_stage: RuntimeStage = RuntimeStage.LOADING,
-        normalize_identifiers: bool = True,
         **kwargs: t.Any,
     ) -> t.List[t.Optional[exp.Expression]]:
         """Renders a expression, expanding macros with provided kwargs
@@ -90,7 +89,6 @@ class BaseExpressionRenderer:
             table_mapping: Table mapping of physical locations. Takes precedence over snapshot mappings.
             deployability_index: Determines snapshots that are deployable in the context of this evaluation.
             runtime_stage: Indicates the current runtime stage, for example if we're still loading the project, etc.
-            normalize_identifiers: Whether or not to normalize and quote identifiers.
             kwargs: Additional kwargs to pass to the renderer.
 
         Returns:
@@ -98,7 +96,7 @@ class BaseExpressionRenderer:
         """
 
         should_cache = self._should_cache(
-            runtime_stage, start, end, execution_time, not normalize_identifiers, *kwargs.values()
+            runtime_stage, start, end, execution_time, *kwargs.values()
         )
 
         if should_cache and self._cache:
@@ -195,7 +193,7 @@ class BaseExpressionRenderer:
                 raise_config_error(f"Failed to resolve macro for expression. {ex}", self._path)
 
             for expression in t.cast(t.List[exp.Expression], transformed_expressions):
-                with self._normalize_and_quote(expression, normalize_identifiers) as expression:
+                with self._normalize_and_quote(expression) as expression:
                     if hasattr(expression, "selects"):
                         for select in expression.selects:
                             if not isinstance(select, exp.Alias) and select.output_name not in (
@@ -297,8 +295,8 @@ class BaseExpressionRenderer:
             return expression
 
     @contextmanager
-    def _normalize_and_quote(self, query: E, normalize_identifiers: bool = True) -> t.Iterator[E]:
-        if self._normalize_identifiers and normalize_identifiers:
+    def _normalize_and_quote(self, query: E) -> t.Iterator[E]:
+        if self._normalize_identifiers:
             with d.normalize_and_quote(
                 query, self._dialect, self._default_catalog, quote=self._quote_identifiers
             ) as query:
@@ -418,7 +416,6 @@ class QueryRenderer(BaseExpressionRenderer):
                     table_mapping=table_mapping,
                     deployability_index=deployability_index,
                     runtime_stage=runtime_stage,
-                    normalize_identifiers=needs_optimization,
                     **kwargs,
                 )
             except ParsetimeAdapterCallError:


Test patch:
diff --git a/pytest.ini b/pytest.ini
index f1731df7..e016a6d0 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -20,6 +20,7 @@ markers =
     spark_pyspark: test for Spark with PySpark dependency
     # Engine Adapters
     engine: test all engine adapters
+    athena: test for Athena
     bigquery: test for BigQuery
     clickhouse: test for Clickhouse (standalone mode)
     clickhouse_cluster: test for Clickhouse (cluster mode)
diff --git a/sqlmesh/core/test/definition.py b/sqlmesh/core/test/definition.py
index 5db053aa..41f40bef 100644
--- a/sqlmesh/core/test/definition.py
+++ b/sqlmesh/core/test/definition.py
@@ -297,7 +297,11 @@ class ModelTest(unittest.TestCase):
             path: An optional path to the test definition yaml file.
             preserve_fixtures: Preserve the fixture tables in the testing database, useful for debugging.
         """
-        name = normalize_model_name(body["model"], default_catalog=default_catalog, dialect=dialect)
+        name = body.get("model")
+        if name is None:
+            _raise_error("Missing required 'model' field", path)
+
+        name = normalize_model_name(name, default_catalog=default_catalog, dialect=dialect)
         model = models.get(name)
         if not model:
             _raise_error(f"Model '{name}' was not found", path)
diff --git a/tests/core/engine_adapter/config.yaml b/tests/core/engine_adapter/config.yaml
index 30d4f5bf..973db17e 100644
--- a/tests/core/engine_adapter/config.yaml
+++ b/tests/core/engine_adapter/config.yaml
@@ -153,5 +153,17 @@ gateways:
     state_connection:
       type: duckdb
 
+  inttest_athena:
+    connection:
+      type: athena
+      aws_access_key_id: {{ env_var("AWS_ACCESS_KEY_ID") }}
+      aws_secret_access_key: {{ env_var("AWS_SECRET_ACCESS_KEY") }}
+      region_name: {{ env_var("AWS_REGION") }}
+      work_group: {{ env_var("ATHENA_WORK_GROUP", "primary") }}
+      s3_warehouse_location: {{ env_var("ATHENA_S3_WAREHOUSE_LOCATION", "") }}
+      concurrent_tasks: 1
+    state_connection:
+      type: duckdb
+
 model_defaults:
     dialect: duckdb
diff --git a/tests/core/engine_adapter/test_athena.py b/tests/core/engine_adapter/test_athena.py
new file mode 100644
index 00000000..b7c161be
--- /dev/null
+++ b/tests/core/engine_adapter/test_athena.py
@@ -0,0 +1,321 @@
+import typing as t
+import pytest
+from pytest_mock import MockerFixture
+import pandas as pd
+
+from sqlglot import exp, parse_one
+import sqlmesh.core.dialect as d
+from sqlmesh.core.engine_adapter import AthenaEngineAdapter
+from sqlmesh.core.model import load_sql_based_model
+from sqlmesh.core.model.definition import SqlModel
+from sqlmesh.utils.errors import SQLMeshError
+
+from tests.core.engine_adapter import to_sql_calls
+
+pytestmark = [pytest.mark.athena, pytest.mark.engine]
+
+
+@pytest.fixture
+def adapter(make_mocked_engine_adapter: t.Callable) -> AthenaEngineAdapter:
+    return make_mocked_engine_adapter(AthenaEngineAdapter)
+
+
+@pytest.mark.parametrize(
+    "config_s3_warehouse_location,table_properties,table,expected_location",
+    [
+        # No s3_warehouse_location in config
+        (None, None, exp.to_table("schema.table"), None),
+        (None, {}, exp.to_table("schema.table"), None),
+        (
+            None,
+            {"s3_base_location": exp.Literal.string("s3://some/location/")},
+            exp.to_table("schema.table"),
+            "s3://some/location/table/",
+        ),
+        # Location set to bucket
+        ("s3://bucket", None, exp.to_table("schema.table"), "s3://bucket/schema/table/"),
+        ("s3://bucket", {}, exp.to_table("schema.table"), "s3://bucket/schema/table/"),
+        ("s3://bucket", None, exp.to_table("schema.table"), "s3://bucket/schema/table/"),
+        (
+            "s3://bucket",
+            {"s3_base_location": exp.Literal.string("s3://some/location/")},
+            exp.to_table("schema.table"),
+            "s3://some/location/table/",
+        ),
+        ("s3://bucket", {}, exp.Table(db=exp.Identifier(this="test")), "s3://bucket/test/"),
+        # Location set to bucket with prefix
+        (
+            "s3://bucket/subpath/",
+            None,
+            exp.to_table("schema.table"),
+            "s3://bucket/subpath/schema/table/",
+        ),
+        ("s3://bucket/subpath/", None, exp.to_table("table"), "s3://bucket/subpath/table/"),
+        (
+            "s3://bucket/subpath/",
+            None,
+            exp.to_table("catalog.schema.table"),
+            "s3://bucket/subpath/catalog/schema/table/",
+        ),
+        (
+            "s3://bucket/subpath/",
+            None,
+            exp.Table(db=exp.Identifier(this="test")),
+            "s3://bucket/subpath/test/",
+        ),
+    ],
+)
+def test_table_location(
+    adapter: AthenaEngineAdapter,
+    config_s3_warehouse_location: t.Optional[str],
+    table_properties: t.Optional[t.Dict[str, exp.Expression]],
+    table: exp.Table,
+    expected_location: t.Optional[str],
+) -> None:
+    adapter.s3_warehouse_location = config_s3_warehouse_location
+    if expected_location is None:
+        with pytest.raises(SQLMeshError, match=r"Cannot figure out location for table.*"):
+            adapter._table_location_or_raise(table_properties, table)
+    else:
+        location = adapter._table_location_or_raise(
+            table_properties, table
+        ).this.name  # extract the unquoted location value from the LocationProperty
+        assert location == expected_location
+
+    if table_properties is not None:
+        # this get consumed by _table_location because we dont want it to end up in a TBLPROPERTIES clause
+        assert "s3_base_location" not in table_properties
+
+
+def test_table_location_length(adapter: AthenaEngineAdapter) -> None:
+    with pytest.raises(SQLMeshError, match=r".*cannot be more than 700 characters"):
+        long_path = "foo/bar/" * 100
+        assert len(long_path) > 700
+
+        adapter._table_location(
+            table_properties={"s3_base_location": exp.Literal.string(f"s3://{long_path}")},
+            table=exp.to_table("foo"),
+        )
+
+
+def test_create_schema(adapter: AthenaEngineAdapter) -> None:
+    adapter.create_schema("test")
+
+    adapter.s3_warehouse_location = "s3://base"
+    adapter.create_schema("test")
+
+    assert to_sql_calls(adapter) == [
+        "CREATE SCHEMA IF NOT EXISTS `test`",
+        "CREATE SCHEMA IF NOT EXISTS `test` LOCATION 's3://base/test/'",
+    ]
+
+
+def test_create_table_hive(adapter: AthenaEngineAdapter) -> None:
+    expressions = d.parse(
+        """
+        MODEL (
+            name test_table,
+            kind FULL,
+            partitioned_by (cola, colb),
+            storage_format parquet,
+            physical_properties (
+                s3_base_location = 's3://foo',
+                has_encrypted_data = 'true'
+            )
+        );
+
+        SELECT 1::timestamp AS cola, 2::varchar as colb, 'foo' as colc;
+    """
+    )
+    model: SqlModel = t.cast(SqlModel, load_sql_based_model(expressions))
+
+    adapter.create_table(
+        model.name,
+        columns_to_types=model.columns_to_types_or_raise,
+        table_properties=model.physical_properties,
+        partitioned_by=model.partitioned_by,
+        storage_format=model.storage_format,
+    )
+
+    assert to_sql_calls(adapter) == [
+        "CREATE EXTERNAL TABLE IF NOT EXISTS `test_table` (`colc` STRING) PARTITIONED BY (`cola` TIMESTAMP, `colb` STRING) STORED AS PARQUET LOCATION 's3://foo/test_table/' TBLPROPERTIES ('has_encrypted_data'='true')"
+    ]
+
+
+def test_create_table_iceberg(adapter: AthenaEngineAdapter) -> None:
+    expressions = d.parse(
+        """
+        MODEL (
+            name test_table,
+            kind FULL,
+            partitioned_by (colc, bucket(16, cola)),
+            storage_format parquet,
+            physical_properties (
+                table_type = 'iceberg',
+                s3_base_location = 's3://foo'
+            )
+        );
+
+        SELECT 1::timestamp AS cola, 2::varchar as colb, 'foo' as colc;
+    """
+    )
+    model: SqlModel = t.cast(SqlModel, load_sql_based_model(expressions))
+
+    adapter.create_table(
+        model.name,
+        columns_to_types=model.columns_to_types_or_raise,
+        table_properties=model.physical_properties,
+        partitioned_by=model.partitioned_by,
+        storage_format=model.storage_format,
+    )
+
+    assert to_sql_calls(adapter) == [
+        "CREATE TABLE IF NOT EXISTS `test_table` (`cola` TIMESTAMP, `colb` STRING, `colc` STRING) PARTITIONED BY (`colc`, BUCKET(16, `cola`)) LOCATION 's3://foo/test_table/' TBLPROPERTIES ('table_type'='iceberg', 'format'='parquet')"
+    ]
+
+
+def test_create_table_no_location(adapter: AthenaEngineAdapter) -> None:
+    expressions = d.parse(
+        """
+        MODEL (
+            name test_table,
+            kind FULL
+        );
+
+        SELECT a::int FROM foo;
+    """
+    )
+    model: SqlModel = t.cast(SqlModel, load_sql_based_model(expressions))
+
+    with pytest.raises(SQLMeshError, match=r"Cannot figure out location.*"):
+        adapter.create_table(
+            model.name,
+            columns_to_types=model.columns_to_types_or_raise,
+            table_properties=model.physical_properties,
+        )
+
+    adapter.s3_warehouse_location = "s3://bucket/prefix"
+    adapter.create_table(
+        model.name,
+        columns_to_types=model.columns_to_types_or_raise,
+        table_properties=model.physical_properties,
+    )
+
+    assert to_sql_calls(adapter) == [
+        "CREATE EXTERNAL TABLE IF NOT EXISTS `test_table` (`a` INT) LOCATION 's3://bucket/prefix/test_table/'",
+    ]
+
+
+def test_ctas_hive(adapter: AthenaEngineAdapter):
+    adapter.s3_warehouse_location = "s3://bucket/prefix/"
+
+    adapter.ctas(
+        table_name="foo.bar",
+        columns_to_types={"a": exp.DataType.build("int")},
+        query_or_df=parse_one("select 1", into=exp.Select),
+    )
+
+    assert to_sql_calls(adapter) == [
+        'CREATE TABLE IF NOT EXISTS "foo"."bar" WITH (external_location=\'s3://bucket/prefix/foo/bar/\') AS SELECT CAST("a" AS INTEGER) AS "a" FROM (SELECT 1) AS "_subquery"'
+    ]
+
+
+def test_ctas_iceberg(adapter: AthenaEngineAdapter):
+    adapter.s3_warehouse_location = "s3://bucket/prefix/"
+
+    adapter.ctas(
+        table_name="foo.bar",
+        columns_to_types={"a": exp.DataType.build("int")},
+        query_or_df=parse_one("select 1", into=exp.Select),
+        table_properties={"table_type": exp.Literal.string("iceberg")},
+    )
+
+    assert to_sql_calls(adapter) == [
+        'CREATE TABLE IF NOT EXISTS "foo"."bar" WITH (location=\'s3://bucket/prefix/foo/bar/\', is_external=false, table_type=\'iceberg\') AS SELECT CAST("a" AS INTEGER) AS "a" FROM (SELECT 1) AS "_subquery"'
+    ]
+
+
+def test_ctas_iceberg_no_specific_location(adapter: AthenaEngineAdapter):
+    with pytest.raises(SQLMeshError, match=r"Cannot figure out location.*"):
+        adapter.ctas(
+            table_name="foo.bar",
+            columns_to_types={"a": exp.DataType.build("int")},
+            query_or_df=parse_one("select 1", into=exp.Select),
+            table_properties={"table_type": exp.Literal.string("iceberg")},
+        )
+
+    assert to_sql_calls(adapter) == []
+
+
+def test_replace_query(adapter: AthenaEngineAdapter, mocker: MockerFixture):
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.athena.AthenaEngineAdapter.table_exists", return_value=True
+    )
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.athena.AthenaEngineAdapter._query_table_type",
+        return_value="iceberg",
+    )
+
+    adapter.replace_query(
+        table_name="test",
+        query_or_df=parse_one("select 1 as a", into=exp.Select),
+        columns_to_types={"a": exp.DataType.build("int")},
+        table_properties={},
+    )
+
+    assert to_sql_calls(adapter) == [
+        'DELETE FROM "test" WHERE TRUE',
+        'INSERT INTO "test" ("a") SELECT 1 AS "a"',
+    ]
+
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.athena.AthenaEngineAdapter.table_exists", return_value=False
+    )
+    adapter.cursor.execute.reset_mock()
+
+    adapter.s3_warehouse_location = "s3://foo"
+    adapter.replace_query(
+        table_name="test",
+        query_or_df=parse_one("select 1 as a", into=exp.Select),
+        columns_to_types={"a": exp.DataType.build("int")},
+        table_properties={},
+    )
+
+    # gets recreated as a Hive table because table_exists=False and nothing in the properties indicates it should be Iceberg
+    assert to_sql_calls(adapter) == [
+        'CREATE TABLE IF NOT EXISTS "test" WITH (external_location=\'s3://foo/test/\') AS SELECT CAST("a" AS INTEGER) AS "a" FROM (SELECT 1 AS "a") AS "_subquery"'
+    ]
+
+
+def test_columns(adapter: AthenaEngineAdapter, mocker: MockerFixture):
+    mock = mocker.patch(
+        "pandas.io.sql.read_sql_query",
+        return_value=pd.DataFrame(
+            data=[["col1", "int"], ["col2", "varchar"]], columns=["column_name", "data_type"]
+        ),
+    )
+
+    assert adapter.columns("foo.bar") == {
+        "col1": exp.DataType.build("int"),
+        "col2": exp.DataType.build("varchar"),
+    }
+
+    assert (
+        mock.call_args_list[0][0][0]
+        == """SELECT "column_name", "data_type" FROM "information_schema"."columns" WHERE "table_schema" = 'foo' AND "table_name" = 'bar' ORDER BY "ordinal_position" NULLS FIRST"""
+    )
+
+
+def test_truncate_table(adapter: AthenaEngineAdapter):
+    adapter._truncate_table(exp.to_table("foo.bar"))
+
+    assert to_sql_calls(adapter) == ['DELETE FROM "foo"."bar"']
+
+
+def test_create_state_table(adapter: AthenaEngineAdapter):
+    adapter.s3_warehouse_location = "s3://base"
+    adapter.create_state_table("_snapshots", {"name": exp.DataType.build("varchar")})
+
+    assert to_sql_calls(adapter) == [
+        "CREATE TABLE IF NOT EXISTS `_snapshots` (`name` STRING) LOCATION 's3://base/_snapshots/' TBLPROPERTIES ('table_type'='iceberg')"
+    ]
diff --git a/tests/core/engine_adapter/test_integration.py b/tests/core/engine_adapter/test_integration.py
index c2b3c4c4..22b472d8 100644
--- a/tests/core/engine_adapter/test_integration.py
+++ b/tests/core/engine_adapter/test_integration.py
@@ -16,6 +16,7 @@ from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 from sqlmesh import Config, Context, EngineAdapter
 from sqlmesh.cli.example_project import init_example_project
 from sqlmesh.core.config import load_config_from_paths
+from sqlmesh.core.config.connection import AthenaConnectionConfig
 from sqlmesh.core.dialect import normalize_model_name
 import sqlmesh.core.dialect as d
 from sqlmesh.core.engine_adapter import SparkEngineAdapter, TrinoEngineAdapter
@@ -180,6 +181,13 @@ class TestContext:
             )
         )
 
+    def physical_properties(
+        self, properties_for_dialect: t.Dict[str, t.Dict[str, str | exp.Expression]]
+    ) -> t.Dict[str, exp.Expression]:
+        if props := properties_for_dialect.get(self.dialect):
+            return {k: exp.Literal.string(v) if isinstance(v, str) else v for k, v in props.items()}
+        return {}
+
     def schema(self, schema_name: str, catalog_name: t.Optional[str] = None) -> str:
         return exp.table_name(
             normalize_model_name(
@@ -414,6 +422,13 @@ class TestContext:
         )
         if config_mutator:
             config_mutator(self.gateway, config)
+
+        if "athena" in self.gateway:
+            # Ensure that s3_warehouse_location is propagated
+            config.gateways[
+                self.gateway
+            ].connection.s3_warehouse_location = self.engine_adapter.s3_warehouse_location
+
         self._context = Context(paths=".", config=config, gateway=self.gateway)
         return self._context
 
@@ -687,19 +702,54 @@ def config() -> Config:
                 pytest.mark.xdist_group("engine_integration_clickhouse_cloud"),
             ],
         ),
+        pytest.param(
+            "athena",
+            marks=[
+                pytest.mark.engine,
+                pytest.mark.remote,
+                pytest.mark.athena,
+                pytest.mark.xdist_group("engine_integration_athena"),
+            ],
+        ),
     ]
 )
 def mark_gateway(request) -> t.Tuple[str, str]:
     return request.param, f"inttest_{request.param}"
 
 
+@pytest.fixture(scope="session")
+def run_count(request) -> t.Iterable[int]:
+    count: int = request.config.cache.get("run_count", 0)
+    count += 1
+    yield count
+    request.config.cache.set("run_count", count)
+
+
 @pytest.fixture
-def engine_adapter(mark_gateway: t.Tuple[str, str], config) -> EngineAdapter:
+def engine_adapter(
+    mark_gateway: t.Tuple[str, str], config, testrun_uid, run_count
+) -> EngineAdapter:
     mark, gateway = mark_gateway
     if gateway not in config.gateways:
         # TODO: Once everything is fully setup we want to error if a gateway is not configured that we expect
         pytest.skip(f"Gateway {gateway} not configured")
     connection_config = config.gateways[gateway].connection
+
+    if mark == "athena":
+        connection_config = t.cast(AthenaConnectionConfig, connection_config)
+        # S3 files need to go into a unique location for each test run
+        # This is because DROP TABLE on a Hive table just drops the table from the metastore
+        # The files still exist in S3, so if you CREATE TABLE to the same location, the old data shows back up
+        # This is a problem for any tests like `test_init_project` that use a consistent schema like `sqlmesh_example` between runs
+        # Note that the `testrun_uid` fixture comes from the xdist plugin
+        testrun_path = f"testrun_{testrun_uid}"
+        if current_location := connection_config.s3_warehouse_location:
+            if testrun_path not in current_location:
+                # only add it if its not already there (since this setup code gets called multiple times in a full test run)
+                connection_config.s3_warehouse_location = os.path.join(
+                    current_location, testrun_path, str(run_count)
+                )
+
     engine_adapter = connection_config.create_engine_adapter()
     # Trino: If we batch up the requests then when running locally we get a table not found error after creating the
     # table and then immediately after trying to insert rows into it. There seems to be a delay between when the
@@ -1276,7 +1326,16 @@ def test_merge(ctx: TestContext):
 
     ctx.init()
     table = ctx.table("test_table")
-    ctx.engine_adapter.create_table(table, ctx.columns_to_types)
+
+    table_properties = ctx.physical_properties(
+        {
+            # Athena only supports MERGE on Iceberg tables
+            # And it cant fall back to a logical merge on Hive tables because it cant delete records
+            "athena": {"table_type": "iceberg"}
+        }
+    )
+
+    ctx.engine_adapter.create_table(table, ctx.columns_to_types, table_properties=table_properties)
     input_data = pd.DataFrame(
         [
             {"id": 1, "ds": "2022-01-01"},
@@ -1346,7 +1405,13 @@ def test_scd_type_2_by_time(ctx: TestContext):
     input_schema = {
         k: v for k, v in ctx.columns_to_types.items() if k not in ("valid_from", "valid_to")
     }
-    ctx.engine_adapter.create_table(table, ctx.columns_to_types)
+    table_properties = ctx.physical_properties(
+        {
+            # Athena only supports the operations required for SCD models on Iceberg tables
+            "athena": {"table_type": "iceberg"}
+        }
+    )
+    ctx.engine_adapter.create_table(table, ctx.columns_to_types, table_properties=table_properties)
     input_data = pd.DataFrame(
         [
             {"id": 1, "name": "a", "updated_at": "2022-01-01 00:00:00"},
@@ -1364,6 +1429,7 @@ def test_scd_type_2_by_time(ctx: TestContext):
         execution_time="2023-01-01 00:00:00",
         updated_at_as_valid_from=False,
         columns_to_types=input_schema,
+        table_properties=table_properties,
     )
     results = ctx.get_metadata_results()
     assert len(results.views) == 0
@@ -1424,6 +1490,7 @@ def test_scd_type_2_by_time(ctx: TestContext):
         execution_time="2023-01-05 00:00:00",
         updated_at_as_valid_from=False,
         columns_to_types=input_schema,
+        table_properties=table_properties,
     )
     results = ctx.get_metadata_results()
     assert len(results.views) == 0
@@ -1489,7 +1556,13 @@ def test_scd_type_2_by_column(ctx: TestContext):
     input_schema = {
         k: v for k, v in ctx.columns_to_types.items() if k not in ("valid_from", "valid_to")
     }
-    ctx.engine_adapter.create_table(table, ctx.columns_to_types)
+    table_properties = ctx.physical_properties(
+        {
+            # Athena only supports the operations required for SCD models on Iceberg tables
+            "athena": {"table_type": "iceberg"}
+        }
+    )
+    ctx.engine_adapter.create_table(table, ctx.columns_to_types, table_properties=table_properties)
     input_data = pd.DataFrame(
         [
             {"id": 1, "name": "a", "status": "active"},
@@ -1721,7 +1794,15 @@ def test_truncate_table(ctx: TestContext):
 
     ctx.init()
     table = ctx.table("test_table")
-    ctx.engine_adapter.create_table(table, ctx.columns_to_types)
+
+    table_properties = ctx.physical_properties(
+        {
+            # Athena only supports TRUNCATE (DELETE FROM <table>) on Iceberg tables
+            "athena": {"table_type": "iceberg"}
+        }
+    )
+
+    ctx.engine_adapter.create_table(table, ctx.columns_to_types, table_properties=table_properties)
     input_data = pd.DataFrame(
         [
             {"id": 1, "ds": "2022-01-01"},
@@ -1773,7 +1854,13 @@ def test_sushi(mark_gateway: t.Tuple[str, str], ctx: TestContext):
         ],
         personal_paths=[pathlib.Path("~/.sqlmesh/config.yaml").expanduser()],
     )
-    _, gateway = mark_gateway
+    mark, gateway = mark_gateway
+
+    if mark == "athena":
+        # Ensure that this test is using the same s3_warehouse_location as TestContext (which includes the testrun_id)
+        config.gateways[
+            gateway
+        ].connection.s3_warehouse_location = ctx.engine_adapter.s3_warehouse_location
 
     # clear cache from prior runs
     cache_dir = pathlib.Path("./examples/sushi/.cache")
@@ -1854,6 +1941,13 @@ def test_sushi(mark_gateway: t.Tuple[str, str], ctx: TestContext):
                 "CREATE VIEW raw.demographics ON CLUSTER cluster1 AS SELECT 1 AS customer_id, '00000' AS zip;"
             )
 
+    # Athena needs models that get mutated after creation to be using Iceberg
+    if ctx.dialect == "athena":
+        for model_name in {"sushi.customer_revenue_lifetime"}:
+            context.get_model(model_name).physical_properties["table_type"] = exp.Literal.string(
+                "iceberg"
+            )
+
     plan: Plan = context.plan(
         environment="test_prod",
         start=start,
@@ -2136,7 +2230,14 @@ def test_init_project(ctx: TestContext, mark_gateway: t.Tuple[str, str], tmp_pat
     if config.model_defaults.dialect != ctx.dialect:
         config.model_defaults = config.model_defaults.copy(update={"dialect": ctx.dialect})
 
-    _, gateway = mark_gateway
+    mark, gateway = mark_gateway
+
+    if mark == "athena":
+        # Ensure that this test is using the same s3_warehouse_location as TestContext (which includes the testrun_id)
+        config.gateways[
+            gateway
+        ].connection.s3_warehouse_location = ctx.engine_adapter.s3_warehouse_location
+
     context = Context(paths=tmp_path, config=config, gateway=gateway)
     ctx.engine_adapter = context.engine_adapter
 
@@ -2368,6 +2469,12 @@ def test_batch_size_on_incremental_by_unique_key_model(
     context.upsert_model(
         create_sql_model(name=f"{schema}.seed_model", query=seed_query, kind="FULL")
     )
+
+    physical_properties = ""
+    if ctx.dialect == "athena":
+        # INCREMENTAL_BY_UNIQUE_KEY uses MERGE which is only supported in Athena on Iceberg tables
+        physical_properties = "physical_properties (table_type = 'iceberg'),"
+
     context.upsert_model(
         load_sql_based_model(
             d.parse(
@@ -2377,6 +2484,7 @@ def test_batch_size_on_incremental_by_unique_key_model(
                         unique_key item_id,
                         batch_size 1
                     ),
+                    {physical_properties}
                     start '2020-01-01',
                     end '2020-01-07',
                     cron '@daily'
diff --git a/tests/core/test_connection_config.py b/tests/core/test_connection_config.py
index b5e6d50a..0eecbf10 100644
--- a/tests/core/test_connection_config.py
+++ b/tests/core/test_connection_config.py
@@ -15,6 +15,7 @@ from sqlmesh.core.config.connection import (
     PostgresConnectionConfig,
     SnowflakeConnectionConfig,
     TrinoAuthenticationMethod,
+    AthenaConnectionConfig,
     _connection_config_validator,
 )
 from sqlmesh.utils.errors import ConfigError
@@ -668,3 +669,55 @@ def test_clickhouse(make_config):
 
     assert not config3.use_compression
     assert not config3._static_connection_kwargs["compress"]
+
+
+def test_athena(make_config):
+    config = make_config(type="athena", work_group="primary")
+    assert isinstance(config, AthenaConnectionConfig)
+
+
+def test_athena_s3_staging_dir_or_workgroup(make_config):
+    with pytest.raises(
+        ConfigError, match=r"At least one of work_group or s3_staging_dir must be set"
+    ):
+        config = make_config(type="athena")
+
+    config = make_config(type="athena", s3_staging_dir="s3://foo")
+
+    assert isinstance(config, AthenaConnectionConfig)
+    assert config.work_group is None
+    assert config.s3_staging_dir == "s3://foo/"  # validator appends trailing /
+
+    config = make_config(type="athena", work_group="test")
+
+    assert isinstance(config, AthenaConnectionConfig)
+    assert config.work_group == "test"
+    assert config.s3_staging_dir is None
+
+
+def test_athena_s3_locations_valid(make_config):
+    with pytest.raises(ConfigError, match=r".*must be a s3:// URI.*"):
+        make_config(
+            type="athena", work_group="primary", s3_warehouse_location="hdfs://legacy/location"
+        )
+
+    with pytest.raises(ConfigError, match=r".*must be a s3:// URI.*"):
+        make_config(type="athena", s3_staging_dir="alskdjlskadgj")
+
+    config = make_config(
+        type="athena",
+        s3_staging_dir="s3://bucket/query-results",
+        s3_warehouse_location="s3://bucket/prod/warehouse/",
+    )
+
+    assert isinstance(config, AthenaConnectionConfig)
+    assert config.s3_staging_dir == "s3://bucket/query-results/"
+    assert config.s3_warehouse_location == "s3://bucket/prod/warehouse/"
+
+    config = make_config(
+        type="athena", work_group="primary", s3_staging_dir=None, s3_warehouse_location=None
+    )
+
+    assert isinstance(config, AthenaConnectionConfig)
+    assert config.s3_staging_dir is None
+    assert config.s3_warehouse_location is None
diff --git a/tests/core/test_test.py b/tests/core/test_test.py
index 6f9263ca..0c50b631 100644
--- a/tests/core/test_test.py
+++ b/tests/core/test_test.py
@@ -108,7 +108,7 @@ def full_model_with_two_ctes(request) -> SqlModel:
         renamed AS (
             SELECT id AS fid FROM source
         )
-        SELECT fid FROM renamed;
+        SELECT fid FROM RENAMED;
         """,
         dialect=getattr(request, "param", None),
         default_catalog="memory",