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 (matched case-insensitively) and the namespace is used as
the table schema.

- Emit only TBLPROPERTIES for CREATE TABLE on an S3 Tables catalog:
  LOCATION, ROW FORMAT, SERDEPROPERTIES, and STORED AS are not accepted
  by managed Iceberg tables. Raise clear compile-time errors when the
  table does not declare table_type ICEBERG or specifies an explicit
  awsathena_location, instead of emitting DDL Athena would reject.
- Store connection options in _create_connect_args so subclass dialects
  (pandas, arrow, polars, s3fs) that call it directly also expose
  catalog_name to the DDL compiler.
- Add no-AWS compiler unit tests and AWS-gated E2E tests (CREATE TABLE
  with reflection, partition transform, CTAS with is_external=false)
  with idempotent DROP TABLE IF EXISTS cleanup, 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).
- Cancel superseded workflow runs with a concurrency group: overlapping
  runs stampede the account's Athena concurrent-query quota
  (TooManyRequestsException) and fail each other.

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

.github/workflows/test.yaml

@@ -12,6 +12,12 @@ permissions:
   id-token: write
   contents: read
 
+# Cancel superseded runs: overlapping runs stampede the account's Athena
+# concurrent-query quota (TooManyRequestsException) and fail each other.
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 jobs:
   test:
     uses: ./.github/workflows/test-suite.yaml

docs/sqlalchemy.md

@@ -363,6 +363,63 @@ 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`, `ROW FORMAT`, and `STORED AS` clauses automatically. Tables must
+declare `awsathena_tblproperties={"table_type": "ICEBERG"}` (S3 Tables support only Iceberg), and
+`awsathena_location` must not be set — the dialect raises a compile-time error otherwise.
+
+```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/base.py

@@ -207,8 +207,7 @@ def create_connect_args(self, url: URL) -> tuple[tuple[str], MutableMapping[str,
         #   awsathena+rest://
         #   {aws_access_key_id}:{aws_secret_access_key}@athena.{region_name}.amazonaws.com:443/
         #   {schema_name}?s3_staging_dir={s3_staging_dir}&...
-        self._connect_options = self._create_connect_args(url)
-        return cast(tuple[str], ()), self._connect_options
+        return cast(tuple[str], ()), self._create_connect_args(url)
 
     def _create_connect_args(self, url: URL) -> dict[str, Any]:
         opts: dict[str, Any] = {
@@ -238,6 +237,12 @@ def _create_connect_args(self, url: URL) -> dict[str, Any]:
             opts.update({"result_reuse_enable": bool(strtobool(opts["result_reuse_enable"]))})
         if "result_reuse_minutes" in opts:
             opts.update({"result_reuse_minutes": int(opts["result_reuse_minutes"])})
+        # Store on the dialect so compilers can consult connection options
+        # (e.g. catalog_name for S3 Tables detection). Assigned here rather than
+        # in create_connect_args because subclass dialects call this method
+        # directly and mutate the returned dict afterwards; sharing the same
+        # object keeps _connect_options in sync with their updates.
+        self._connect_options = opts
         return opts
 
     @reflection.cache

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,74 @@ 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 ""
+        # Athena resolves catalog names case-insensitively.
+        return catalog.lower().startswith(S3_TABLES_CATALOG_PREFIX)
+
+    def _is_iceberg_table(
+        self, dialect_opts: _DialectArgDict, connect_opts: Mapping[str, Any]
+    ) -> bool:
+        """Return whether the table properties declare an Iceberg table.
+
+        Args:
+            dialect_opts: The table's ``awsathena_*`` dialect options.
+            connect_opts: The dialect connection options.
+
+        Returns:
+            True if the rendered TBLPROPERTIES set ``table_type`` to Iceberg.
+        """
+        table_properties = self._get_table_properties_specification(
+            dialect_opts, connect_opts
+        ).lower()
+        return ("table_type" in table_properties) and ("iceberg" in table_properties)
+
+    def _validate_s3_tables_create_table(
+        self, dialect_opts: _DialectArgDict, connect_opts: Mapping[str, Any]
+    ) -> None:
+        """Validate a CREATE TABLE compiled against an S3 Tables catalog.
+
+        S3 Tables support only Iceberg tables on managed storage, so the table
+        must declare ``table_type`` ICEBERG and must not specify a location.
+        Raising here surfaces a clear client-side error instead of emitting DDL
+        that Athena would reject.
+
+        Args:
+            dialect_opts: The table's ``awsathena_*`` dialect options.
+            connect_opts: The dialect connection options.
+
+        Raises:
+            exc.CompileError: If the table is not Iceberg or specifies a location.
+        """
+        if not self._is_iceberg_table(dialect_opts, connect_opts):
+            raise exc.CompileError(
+                "S3 Tables support only Iceberg tables; specify the dialect keyword "
+                "argument `awsathena_tblproperties={'table_type': 'ICEBERG'}`"
+            )
+        if dialect_opts["location"]:
+            raise exc.CompileError(
+                "S3 Tables use managed storage and do not accept a table location; "
+                "remove the dialect keyword argument `awsathena_location`"
+            )
+
     def _get_table_location(
         self, table: Table, dialect_opts: _DialectArgDict, connect_opts: Mapping[str, Any]
     ) -> str | None:
@@ -662,12 +741,7 @@ def visit_create_table(self, create: CreateTable, **kwargs) -> str:
         dialect = cast("AthenaDialect", self.dialect)
         connect_opts = dialect._connect_options
 
-        table_properties = self._get_table_properties_specification(
-            dialect_opts, connect_opts
-        ).lower()
-        is_iceberg = False
-        if ("table_type" in table_properties) and ("iceberg" in table_properties):
-            is_iceberg = True
+        is_iceberg = self._is_iceberg_table(dialect_opts, connect_opts)
 
         # https://docs.aws.amazon.com/athena/latest/ug/querying-iceberg-creating-tables.html
         text = ["\nCREATE TABLE"] if is_iceberg else ["\nCREATE EXTERNAL TABLE"]
@@ -705,11 +779,19 @@ def post_create_table(self, table: Table) -> str:
         dialect_opts: _DialectArgDict = table.dialect_options["awsathena"]
         dialect = cast("AthenaDialect", self.dialect)
         connect_opts = dialect._connect_options
-        text = [
-            self._get_row_format_specification(dialect_opts, connect_opts),
-            self._get_serde_properties_specification(dialect_opts, connect_opts),
-            self._get_file_format_specification(dialect_opts, connect_opts),
-            self._get_table_location_specification(table, dialect_opts, connect_opts),
-            self._get_table_properties_specification(dialect_opts, connect_opts),
-        ]
+        if self._is_s3_tables_catalog(connect_opts):
+            # S3 Tables are managed Iceberg tables: ROW FORMAT, SERDEPROPERTIES,
+            # STORED AS, and LOCATION are not accepted, so emit only TBLPROPERTIES.
+            self._validate_s3_tables_create_table(dialect_opts, connect_opts)
+            text = [
+                self._get_table_properties_specification(dialect_opts, connect_opts),
+            ]
+        else:
+            text = [
+                self._get_row_format_specification(dialect_opts, connect_opts),
+                self._get_serde_properties_specification(dialect_opts, connect_opts),
+                self._get_file_format_specification(dialect_opts, connect_opts),
+                self._get_table_location_specification(table, dialect_opts, connect_opts),
+                self._get_table_properties_specification(dialect_opts, connect_opts),
+            ]
         return "\n".join([t for t in text if t])

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,83 @@ 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.
+  # NOTE: Because the bucket is retained, recreating the stack from scratch will
+  # fail with an AlreadyExists error on this resource; re-import it instead
+  # (create-change-set --change-set-type IMPORT --resources-to-import ...).
+  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)
         )