Backport bug fixes and the CI test policy to 3.x for v3.36.1

.github/workflows/release.yaml

@@ -5,13 +5,23 @@ on:
     tags:
       - 'v*'
 
-permissions:
-  id-token: write
-  contents: write
+permissions: {}
 
 jobs:
+  # Runs every suite on every supported Python version for the tagged commit;
+  # nothing is built or published unless all of them pass.
+  test:
+    uses: ./.github/workflows/test.yaml
+    permissions:
+      contents: read
+      id-token: write
+
   release:
+    needs: test
     runs-on: ubuntu-latest
+    permissions:
+      id-token: write
+      contents: write
 
     env:
       PYTHON_VERSION: '3.12'
@@ -22,7 +32,7 @@ jobs:
 
       - uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
         with:
-          python-version: ${{ matrix.python-version }}
+          python-version: ${{ env.PYTHON_VERSION }}
           enable-cache: true
 
       - name: Build

.github/workflows/test-suite.yaml

@@ -6,9 +6,15 @@ on:
       test-type:
         required: true
         type: string
+      python-versions:
+        description: JSON array of the Python versions to test
+        required: true
+        type: string
 
 jobs:
   run:
+    # External fork contributions must validate AWS behavior in their own account.
+    if: github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository
     runs-on: ubuntu-latest
 
     env:
@@ -18,16 +24,15 @@ 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.
+      # The SQLAlchemy S3 Tables tests need a fixed namespace, which the test
+      # account no longer provides (master creates one per test session), so
+      # they are skipped on this branch.
       AWS_ATHENA_S3_TABLES_CATALOG: s3tablescatalog/laughingman7743-pyathena-s3-tables
-      AWS_ATHENA_S3_TABLES_NAMESPACE: pyathena
 
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
+        python-version: ${{ fromJSON(inputs.python-versions) }}
 
     steps:
       - name: Checkout

.github/workflows/test.yaml

@@ -2,15 +2,23 @@ name: Test
 
 on:
   pull_request:
+    # ready_for_review starts the AWS jobs for a pull request leaving Draft;
+    # converted_to_draft starts a run without them, which cancels an
+    # in-progress run through the concurrency group.
+    types: [opened, synchronize, reopened, ready_for_review, converted_to_draft]
     paths-ignore:
       - 'docs/**'
       - '**.md'
-  schedule:
-    - cron:  '0 0 * * 0'
-  # Allows refreshing the README status badge on demand: the badge reflects
-  # the latest run on the default branch, which is otherwise only the weekly
-  # scheduled run and stays red for up to a week after a transient failure.
+  # Runs every suite on the selected branch, on the requested Python versions.
   workflow_dispatch:
+    inputs:
+      python-versions:
+        description: Comma-separated Python versions, such as 3.12 or 3.11,3.14; empty for every supported version
+        type: string
+        default: ''
+  # The Release workflow runs every suite on every supported Python version
+  # before publishing.
+  workflow_call:
 
 permissions:
   id-token: write
@@ -23,19 +31,83 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
+  # Offline checks run for every event, including Draft and fork pull requests.
+  lint:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+    steps:
+      - name: Checkout
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@37802adc94f370d6bfd71619e3f0bf239e1f3b78 # v7.6.0
+        with:
+          python-version: '3.12'
+          enable-cache: true
+      - uses: taiki-e/install-action@7a79fe8c3a13344501c80d99cae481c1c9085912 # v2.81.10
+        with:
+          tool: just
+      - run: just lint
+
+  # Selects the Python versions of the AWS suites. Draft and external-fork pull
+  # requests run none. A ready pull request tests the newest Python version; a
+  # dispatch tests the requested versions or every version, and the Release
+  # workflow every version.
+  versions:
+    if: >-
+      github.event_name != 'pull_request' ||
+      (!github.event.pull_request.draft &&
+       github.event.pull_request.head.repo.full_name == github.repository)
+    runs-on: ubuntu-latest
+    permissions: {}
+    outputs:
+      python-versions: ${{ steps.select.outputs.python-versions }}
+    steps:
+      - id: select
+        env:
+          EVENT_NAME: ${{ github.event_name }}
+          REQUESTED_VERSIONS: ${{ inputs.python-versions }}
+          # Every supported version, oldest first; keep in sync with the
+          # pyproject.toml classifiers.
+          PYTHON_VERSIONS: '["3.10", "3.11", "3.12", "3.13", "3.14"]'
+        run: |
+          case "$EVENT_NAME" in
+            pull_request)
+              versions=$(jq -c '[last]' <<< "$PYTHON_VERSIONS")
+              ;;
+            workflow_dispatch)
+              versions=$(jq -c --arg requested "$REQUESTED_VERSIONS" '
+                ($requested | split(",") | map(gsub("\\s"; "")) | map(select(. != "")) | unique) as $selected
+                | if $selected == [] then .
+                  elif ($selected - .) == [] then $selected
+                  else error("unsupported Python versions: \($selected - . | join(", "))")
+                  end' <<< "$PYTHON_VERSIONS")
+              ;;
+            *)
+              # The Release workflow (a workflow_call from a tag push).
+              versions=$(jq -c '.' <<< "$PYTHON_VERSIONS")
+              ;;
+          esac
+          echo "python-versions=$versions" >> "$GITHUB_OUTPUT"
+
   test:
+    needs: versions
     uses: ./.github/workflows/test-suite.yaml
     with:
       test-type: pyathena
+      python-versions: ${{ needs.versions.outputs.python-versions }}
 
   test-sqla:
-    needs: [test]
+    needs: [versions, test]
     uses: ./.github/workflows/test-suite.yaml
     with:
       test-type: sqla
+      python-versions: ${{ needs.versions.outputs.python-versions }}
 
   test-sqla-async:
-    needs: [test-sqla]
+    needs: [versions, test-sqla]
     uses: ./.github/workflows/test-suite.yaml
     with:
       test-type: sqla_async
+      python-versions: ${{ needs.versions.outputs.python-versions }}

docs/sqlalchemy.md

@@ -697,12 +697,14 @@ This generates the following SQL structure:
 
 ```sql
 CREATE TABLE users (
-    id INTEGER,
-    profile ROW(name STRING, age INTEGER, email STRING),
-    settings ROW(theme STRING, notifications ROW(email STRING, push STRING))
+    id INT,
+    profile STRUCT<name:STRING, age:INTEGER, email:STRING>,
+    settings STRUCT<theme:STRING, notifications:STRUCT<email:STRING, push:STRING>>
 )
 ```
 
+`CREATE TABLE` renders `AthenaStruct` columns with Hive `STRUCT<name:type, ...>` syntax at every nesting depth, including STRUCT values inside MAP and ARRAY.
+
 #### Querying STRUCT data
 
 PyAthena automatically converts STRUCT data between different formats:

pyathena/pandas/util.py

@@ -261,13 +261,17 @@ def to_sql(
     ).Bucket(bucket_name)
     cursor = conn.cursor()
 
+    # Athena stores identifiers in lowercase and information_schema reports them
+    # that way, so compare lowercase literals.
+    schema_literal = schema.lower().replace("'", "''")
+    name_literal = name.lower().replace("'", "''")
     table = cursor.execute(
         textwrap.dedent(
             f"""
             SELECT table_name
             FROM information_schema.tables
-            WHERE table_schema = '{schema}'
-            AND table_name = '{name}'
+            WHERE table_schema = '{schema_literal}'
+            AND table_name = '{name_literal}'
             """
         )
     ).fetchall()

pyathena/spark/async_cursor.py

@@ -75,6 +75,27 @@ def __init__(
         max_workers: int = (cpu_count() or 1) * 5,
         **kwargs,
     ):
+        """Initialize the cursor and start or attach to a Spark session.
+
+        Args:
+            session_id: ID of an existing session to use. If omitted, a new
+                session is started.
+            description: Description of a new session.
+            engine_configuration: Engine configuration of a new session.
+            notebook_version: Notebook version of a new session.
+            session_idle_timeout_minutes: Idle timeout of a new session in minutes.
+            max_workers: Maximum number of threads for asynchronous operations.
+            **kwargs: Arguments passed to ``SparkBaseCursor``.
+
+        Raises:
+            ValueError: If ``max_workers`` is not greater than 0.
+            OperationalError: If the supplied session does not exist, or the
+                session cannot be started or does not become idle.
+        """
+        # Created before the session so that an invalid max_workers cannot leave
+        # a newly started session behind; the executor starts no threads until used.
+        self._max_workers = max_workers
+        self._executor = ThreadPoolExecutor(max_workers=max_workers)
         super().__init__(
             session_id=session_id,
             description=description,
@@ -83,12 +104,24 @@ def __init__(
             session_idle_timeout_minutes=session_idle_timeout_minutes,
             **kwargs,
         )
-        self._max_workers = max_workers
-        self._executor = ThreadPoolExecutor(max_workers=max_workers)
 
     def close(self, wait: bool = False) -> None:
-        super().close()
-        self._executor.shutdown(wait=wait)
+        """Terminate the Spark session, then shut down the executor.
+
+        The executor is shut down even if terminating the session fails.
+        If termination fails, calling this method again retries it.
+
+        Args:
+            wait: Whether to wait for submitted futures to finish before returning
+                or raising.
+
+        Raises:
+            OperationalError: If terminating the session fails.
+        """
+        try:
+            super().close()
+        finally:
+            self._executor.shutdown(wait=wait)
 
     def calculation_execution(self, query_id: str) -> "Future[AthenaCalculationExecution]":
         return self._executor.submit(self._get_calculation_execution, query_id)