Support Amazon S3 Tables in the SQLAlchemy dialect

Athena exposes S3 Tables through a catalog named
s3tablescatalog/<table-bucket> whose namespace is the database, and such
tables use managed storage. Athena rejects a three-part
catalog.namespace.table identifier in DDL ("Unsupported ddl with 2
catalogs"), so the catalog is selected on the connection via
catalog_name and the namespace is used as the table schema.

- Omit the LOCATION clause for Iceberg tables when the connection
  catalog_name names an S3 Tables catalog (managed storage), instead of
  raising or building an s3_staging_dir fallback.
- Add no-AWS compiler unit tests and AWS-gated E2E tests (CREATE TABLE,
  partition transform, CTAS with is_external=false), an S3 Tables docs
  section, and optional AWS_ATHENA_S3_TABLES_* test env vars.
- Provision the S3 Tables test infrastructure in CloudFormation: table
  bucket, namespace, the federated s3tablescatalog Glue catalog
  (IAM-based access), OIDC-role IAM permissions, and 1-day unreferenced
  file removal to bound storage cost. Wire the env vars into the test
  workflow.
- Import the existing S3 staging bucket into the CloudFormation stack
  with DeletionPolicy: Retain, capturing its lifecycle, encryption,
  ownership, and public-access settings (drift-free import).

Refs #710

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: laughingman7743

.github/workflows/test-suite.yaml

@@ -18,6 +18,11 @@ jobs:
       AWS_ATHENA_WORKGROUP: pyathena
       AWS_ATHENA_SPARK_WORKGROUP: pyathena-spark
       AWS_ATHENA_MANAGED_WORKGROUP: pyathena-managed
+      # Registered S3 Tables catalog (s3tablescatalog/<table-bucket>) and namespace
+      # for the SQLAlchemy S3 Tables tests; the table bucket, namespace, and the
+      # AWS analytics-services integration are provisioned out of band.
+      AWS_ATHENA_S3_TABLES_CATALOG: s3tablescatalog/laughingman7743-pyathena-s3-tables
+      AWS_ATHENA_S3_TABLES_NAMESPACE: pyathena
 
     strategy:
       fail-fast: false

docs/sqlalchemy.md

@@ -363,6 +363,61 @@ If you want to limit the column options to specific table names only, specify th
 awsathena+rest://:@athena.us-west-2.amazonaws.com:443/default?partition=table1.column1%2Ctable1.column2&cluster=table2.column1%2Ctable2.column2&...
 ```
 
+## Amazon S3 Tables
+
+[Amazon S3 Tables](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables.html) are Iceberg-backed
+tables stored in a dedicated table bucket. Once the table bucket is integrated with the AWS analytics services,
+Athena registers it as a catalog named `s3tablescatalog/<table-bucket>`, with the S3 Tables namespace as the
+database.
+
+Athena does not accept a three-part `catalog.namespace.table` identifier in DDL, so select the catalog on the
+**connection** with `catalog_name` and use the namespace as the table `schema`. Because S3 Tables use managed
+storage, the dialect omits the `LOCATION` clause automatically (so `awsathena_location` is not required).
+
+```python
+engine = create_engine(
+    "awsathena+rest://athena.us-west-2.amazonaws.com:443/my_namespace"
+    "?s3_staging_dir=s3://my-bucket/athena-results/"
+    "&catalog_name=s3tablescatalog/my-table-bucket"
+)
+
+table = Table(
+    "some_table",
+    MetaData(schema="my_namespace"),
+    Column("id", types.Integer),
+    Column("dt", types.Date, awsathena_partition=True, awsathena_partition_transform="day"),
+    awsathena_tblproperties={"table_type": "ICEBERG"},
+)
+with engine.connect() as conn:
+    table.create(bind=conn)
+```
+
+which builds the following statement:
+
+```sql
+CREATE TABLE my_namespace.some_table (
+  id INT,
+  dt DATE
+)
+PARTITIONED BY (
+  day(dt)
+)
+TBLPROPERTIES (
+  'table_type' = 'ICEBERG'
+)
+```
+
+All Iceberg partition transforms (`year`, `month`, `day`, `hour`, `bucket`, `truncate`) are supported, the same as
+for other Iceberg tables. CTAS (`CREATE TABLE ... AS SELECT`) is not modeled as a SQLAlchemy construct; issue it as
+raw SQL via `text()`. Managed Iceberg CTAS requires `is_external = false`:
+
+```python
+conn.execute(text(
+    'CREATE TABLE "my_namespace"."some_table" '
+    "WITH (table_type = 'ICEBERG', is_external = false) AS SELECT 1 AS id"
+))
+```
+
 ## Temporal/Time-travel with Iceberg
 
 Athena supports time-travel queries on Iceberg tables by either a version_id or a timestamp. The `FOR TIMESTAMP AS OF`

pyathena/sqlalchemy/compiler.py

@@ -39,6 +39,13 @@
     _DialectArgDict = Mapping[str, Any]
     CreateColumn = Any
 
+# Prefix of the Athena data catalog name registered for an Amazon S3 Tables
+# table bucket (e.g. ``s3tablescatalog/my-bucket``). It is selected via the
+# connection ``catalog_name``. S3 Tables are Iceberg-backed and use managed
+# storage, so their CREATE TABLE statements must not include a LOCATION clause.
+# https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-tables-integrations-query-athena.html
+S3_TABLES_CATALOG_PREFIX = "s3tablescatalog/"
+
 
 class AthenaTypeCompiler(GenericTypeCompiler):
     """Type compiler for Amazon Athena SQL types.
@@ -305,6 +312,10 @@ class AthenaDDLCompiler(DDLCompiler):
 
     - External table creation (EXTERNAL keyword for Hive-style tables)
     - Iceberg table creation (managed tables with ACID support)
+    - Amazon S3 Tables (Iceberg-backed, managed storage): set the connection
+      ``catalog_name`` to ``s3tablescatalog/<table-bucket>`` and use the
+      namespace as the table ``schema``. The LOCATION clause is omitted since
+      storage is managed.
     - File formats: PARQUET, ORC, TEXTFILE, JSON, AVRO, etc.
     - Row formats with SerDe specifications
     - Compression settings for various file formats
@@ -445,6 +456,28 @@ def _get_serde_properties_specification(
             text.append(")")
         return "\n".join(text)
 
+    @staticmethod
+    def _is_s3_tables_catalog(connect_opts: Mapping[str, Any]) -> bool:
+        """Return whether the connection targets an Amazon S3 Tables catalog.
+
+        S3 Tables are queried by setting the connection ``catalog_name`` to
+        ``s3tablescatalog/<table-bucket>`` and using the namespace as the table
+        ``schema`` (a two-part ``namespace.table`` identifier). Athena rejects a
+        three-part ``catalog.namespace.table`` identifier in DDL, so the catalog
+        must be selected at the connection level. Such tables use managed
+        storage, so their CREATE TABLE statement must omit the LOCATION clause.
+
+        Args:
+            connect_opts: The dialect connection options.
+
+        Returns:
+            True if ``catalog_name`` names an S3 Tables catalog.
+        """
+        if not connect_opts:
+            return False
+        catalog = connect_opts.get("catalog_name") or ""
+        return catalog.startswith(S3_TABLES_CATALOG_PREFIX)
+
     def _get_table_location(
         self, table: Table, dialect_opts: _DialectArgDict, connect_opts: Mapping[str, Any]
     ) -> str | None:
@@ -466,6 +499,9 @@ def _get_table_location(
     def _get_table_location_specification(
         self, table: Table, dialect_opts: _DialectArgDict, connect_opts: Mapping[str, Any]
     ) -> str:
+        if self._is_s3_tables_catalog(connect_opts):
+            # S3 Tables use managed storage; CREATE TABLE must not set a LOCATION.
+            return ""
         location = self._get_table_location(table, dialect_opts, connect_opts)
         text = []
         if location:

scripts/cloudformation/github_actions_oidc.yaml

@@ -20,6 +20,14 @@ Parameters:
   ManagedWorkGroupName:
     Type: String
     Default: "pyathena-managed"
+  S3TablesBucketName:
+    Type: String
+    Default: "laughingman7743-pyathena-s3-tables"
+    Description: Name of the Amazon S3 Tables table bucket used by the SQLAlchemy S3 Tables tests.
+  S3TablesNamespaceName:
+    Type: String
+    Default: "pyathena"
+    Description: Namespace created in the S3 Tables table bucket for the SQLAlchemy S3 Tables tests.
   OIDCProviderArn:
     Type: String
     Default: ""
@@ -120,6 +128,30 @@ Resources:
                 Resource: [
                   !Sub "arn:aws:s3:::*"
                 ]
+        - PolicyName: s3tables-access
+          PolicyDocument:
+            Version: "2012-10-17"
+            Statement:
+              # Amazon S3 Tables data plane: create/query/drop namespaces and
+              # tables in the table bucket used by the SQLAlchemy S3 Tables tests.
+              - Effect: Allow
+                Action: [
+                  "s3tables:*"
+                ]
+                Resource: [
+                  !Sub "arn:aws:s3tables:${AWS::Region}:${AWS::AccountId}:bucket/*"
+                ]
+              # Resolve the federated `s3tablescatalog` Glue catalog and obtain
+              # Lake Formation data access when querying S3 Tables from Athena.
+              - Effect: Allow
+                Action: [
+                  "glue:GetCatalog",
+                  "glue:GetCatalogs",
+                  "lakeformation:GetDataAccess"
+                ]
+                Resource: [
+                  "*"
+                ]
   SparkRole:
     Type: AWS::IAM::Role
     Properties:
@@ -269,6 +301,80 @@ Resources:
         ManagedQueryResultsConfiguration:
           Enabled: true
 
+  # S3 staging bucket for Athena query results and test data. Created outside
+  # CloudFormation and imported into the stack. Retain on delete/replace so the
+  # bucket and its data survive stack operations.
+  S3StagingBucket:
+    Type: AWS::S3::Bucket
+    DeletionPolicy: Retain
+    UpdateReplacePolicy: Retain
+    Properties:
+      BucketName: !Sub "${BucketName}"
+      PublicAccessBlockConfiguration:
+        BlockPublicAcls: true
+        IgnorePublicAcls: true
+        BlockPublicPolicy: true
+        RestrictPublicBuckets: true
+      OwnershipControls:
+        Rules:
+          - ObjectOwnership: BucketOwnerEnforced
+      BucketEncryption:
+        ServerSideEncryptionConfiguration:
+          - ServerSideEncryptionByDefault:
+              SSEAlgorithm: AES256
+            BucketKeyEnabled: true
+      LifecycleConfiguration:
+        Rules:
+          - Id: delete
+            Status: Enabled
+            ExpirationInDays: 1
+            NoncurrentVersionExpiration:
+              NoncurrentDays: 1
+              NewerNoncurrentVersions: 1
+            AbortIncompleteMultipartUpload:
+              DaysAfterInitiation: 1
+
+  # Amazon S3 Tables infrastructure for the SQLAlchemy S3 Tables tests.
+  # UnreferencedFileRemoval garbage-collects files left behind by dropped test
+  # tables; 1 day is the most aggressive setting to keep test storage costs low.
+  S3TablesBucket:
+    Type: AWS::S3Tables::TableBucket
+    Properties:
+      TableBucketName: !Sub "${S3TablesBucketName}"
+      UnreferencedFileRemoval:
+        Status: Enabled
+        UnreferencedDays: 1
+        NoncurrentDays: 1
+
+  S3TablesNamespace:
+    Type: AWS::S3Tables::Namespace
+    Properties:
+      TableBucketARN: !GetAtt S3TablesBucket.TableBucketARN
+      Namespace: !Sub "${S3TablesNamespaceName}"
+
+  # One-time, per-Region integration: register S3 Tables into the Glue Data
+  # Catalog as the federated `s3tablescatalog` catalog so Athena can query them.
+  # IAM_ALLOWED_PRINCIPALS default permissions select IAM-based access control
+  # (no Lake Formation grants needed; the OIDC role's IAM policy governs access).
+  S3TablesCatalog:
+    Type: AWS::Glue::Catalog
+    Properties:
+      Name: s3tablescatalog
+      Description: Federated catalog that maps S3 Tables into the Glue Data Catalog
+      FederatedCatalog:
+        Identifier: !Sub "arn:aws:s3tables:${AWS::Region}:${AWS::AccountId}:bucket/*"
+        ConnectionName: aws:s3tables
+      CreateDatabaseDefaultPermissions:
+        - Principal:
+            DataLakePrincipalIdentifier: IAM_ALLOWED_PRINCIPALS
+          Permissions:
+            - ALL
+      CreateTableDefaultPermissions:
+        - Principal:
+            DataLakePrincipalIdentifier: IAM_ALLOWED_PRINCIPALS
+          Permissions:
+            - ALL
+
   GithubOidc:
     Type: AWS::IAM::OIDCProvider
     Condition: CreateOIDCProvider

tests/__init__.py

@@ -31,6 +31,13 @@ def __init__(self):
         )
         self.default_work_group = os.getenv("AWS_ATHENA_DEFAULT_WORKGROUP", "primary")
         self.managed_work_group = os.getenv("AWS_ATHENA_MANAGED_WORKGROUP")
+        # Optional Amazon S3 Tables configuration for the SQLAlchemy dialect tests.
+        # `s3tables_catalog` is the registered table-bucket catalog, e.g.
+        # "s3tablescatalog/<table-bucket>"; `s3tables_namespace` is a namespace that
+        # already exists in it. These are optional so the suite stays runnable
+        # without an S3 Tables bucket; the S3 Tables tests skip when they are unset.
+        self.s3tables_catalog = os.getenv("AWS_ATHENA_S3_TABLES_CATALOG")
+        self.s3tables_namespace = os.getenv("AWS_ATHENA_S3_TABLES_NAMESPACE")
         self.schema = "pyathena_test_" + "".join(
             random.choices(string.ascii_lowercase + string.digits, k=10)
         )

tests/pyathena/conftest.py

@@ -76,6 +76,7 @@ def create_engine(**kwargs):
     conn_str = SQLALCHEMY_CONNECTION_STRING.replace("+rest", f"+{driver}")
     for arg in [
         "bucket_count",
+        "catalog_name",
         "cluster",
         "compression",
         "duration_seconds",