Add S3Executor abstraction for pluggable parallel execution

Introduce S3Executor ABC with two implementations:
- S3ThreadPoolExecutor: wraps ThreadPoolExecutor (default for sync)
- S3AioExecutor: dispatches work via asyncio event loop

S3Executor supports the context manager protocol for safe resource
cleanup. S3FileSystem methods now use `with` statements instead of
manual try/finally for short-lived executors.

Replace hardcoded ThreadPoolExecutor usage in S3File and S3FileSystem
with the S3Executor interface. This enables async-aware parallel
operations without thread-in-thread nesting.

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

Author: laughingman7743

pyathena/filesystem/s3.py

@@ -1,13 +1,11 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
-import itertools
 import logging
 import mimetypes
 import os.path
 import re
 from concurrent.futures import Future, as_completed
-from concurrent.futures.thread import ThreadPoolExecutor
 from copy import deepcopy
 from datetime import datetime
 from multiprocessing import cpu_count
@@ -23,6 +21,7 @@
 from fsspec.utils import tokenize
 
 import pyathena
+from pyathena.filesystem.s3_executor import S3Executor, S3ThreadPoolExecutor
 from pyathena.filesystem.s3_object import (
     S3CompleteMultipartUpload,
     S3MultipartUpload,
@@ -686,6 +685,20 @@ def _delete_object(
             **request,
         )
 
+    def _create_executor(self, max_workers: int) -> S3Executor:
+        """Create an executor strategy for parallel operations.
+
+        Subclasses can override to provide alternative execution strategies
+        (e.g., asyncio-based execution).
+
+        Args:
+            max_workers: Maximum number of parallel workers.
+
+        Returns:
+            An S3Executor instance.
+        """
+        return S3ThreadPoolExecutor(max_workers=max_workers)
+
     def _delete_objects(
         self, bucket: str, paths: List[str], max_workers: Optional[int] = None, **kwargs
     ) -> None:
@@ -703,7 +716,7 @@ def _delete_objects(
                     object_.update({"VersionId": version_id})
                 delete_objects.append(object_)
 
-        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        with self._create_executor(max_workers=max_workers) as executor:
             fs = []
             for delete in [
                 delete_objects[i : i + self.DELETE_OBJECTS_MAX_KEYS]
@@ -861,7 +874,7 @@ def _copy_object_with_multipart_upload(
             **kwargs,
         )
         parts = []
-        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        with self._create_executor(max_workers=max_workers) as executor:
             fs = [
                 executor.submit(
                     self._upload_part_copy,
@@ -1106,6 +1119,7 @@ def _open(
             mode,
             version_id=None,
             max_workers=max_workers,
+            executor=self._create_executor(max_workers=max_workers),
             block_size=block_size,
             cache_type=cache_type,
             autocommit=autocommit,
@@ -1256,6 +1270,7 @@ def __init__(
         mode: str = "rb",
         version_id: Optional[str] = None,
         max_workers: int = (cpu_count() or 1) * 5,
+        executor: Optional[S3Executor] = None,
         block_size: int = S3FileSystem.DEFAULT_BLOCK_SIZE,
         cache_type: str = "bytes",
         autocommit: bool = True,
@@ -1265,7 +1280,7 @@ def __init__(
         **kwargs,
     ) -> None:
         self.max_workers = max_workers
-        self._executor = ThreadPoolExecutor(max_workers=max_workers)
+        self._executor: S3Executor = executor or S3ThreadPoolExecutor(max_workers=max_workers)
         self.s3_additional_kwargs = s3_additional_kwargs if s3_additional_kwargs else {}
 
         super().__init__(
@@ -1481,24 +1496,18 @@ def _fetch_range(self, start: int, end: int) -> bytes:
             start, end, max_workers=self.max_workers, worker_block_size=self.blocksize
         )
         if len(ranges) > 1:
-            object_ = self._merge_objects(
-                list(
-                    self._executor.map(
-                        lambda bucket, key, ranges, version_id, kwargs: self.fs._get_object(
-                            bucket=bucket,
-                            key=key,
-                            ranges=ranges,
-                            version_id=version_id,
-                            **kwargs,
-                        ),
-                        itertools.repeat(self.bucket),
-                        itertools.repeat(self.key),
-                        ranges,
-                        itertools.repeat(self.version_id),
-                        itertools.repeat(self.s3_additional_kwargs),
-                    )
+            futures = [
+                self._executor.submit(
+                    self.fs._get_object,
+                    bucket=self.bucket,
+                    key=self.key,
+                    ranges=r,
+                    version_id=self.version_id,
+                    **self.s3_additional_kwargs,
                 )
-            )
+                for r in ranges
+            ]
+            object_ = self._merge_objects([f.result() for f in as_completed(futures)])
         else:
             object_ = self.fs._get_object(
                 self.bucket,

pyathena/filesystem/s3_executor.py

@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+
+import asyncio
+from abc import ABCMeta, abstractmethod
+from concurrent.futures import Future
+from concurrent.futures.thread import ThreadPoolExecutor
+from typing import Any, Callable, Optional, TypeVar
+
+T = TypeVar("T")
+
+
+class S3Executor(metaclass=ABCMeta):
+    """Abstract executor for parallel S3 operations.
+
+    Defines the interface used by ``S3File`` and ``S3FileSystem`` for submitting
+    work to run in parallel and for shutting down the executor when done.
+    Both ``submit`` and ``shutdown`` mirror the ``concurrent.futures.Executor``
+    interface so that ``as_completed()`` and ``Future.cancel()`` work unchanged.
+    """
+
+    @abstractmethod
+    def submit(self, fn: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]:
+        """Submit a callable for execution and return a Future."""
+        ...
+
+    @abstractmethod
+    def shutdown(self, wait: bool = True) -> None:
+        """Shut down the executor, freeing any resources."""
+        ...
+
+    def __enter__(self) -> "S3Executor":
+        return self
+
+    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        self.shutdown(wait=True)
+
+
+class S3ThreadPoolExecutor(S3Executor):
+    """Executor that delegates to a ``ThreadPoolExecutor``.
+
+    This is the default executor used by ``S3File`` and ``S3FileSystem``
+    for synchronous parallel operations.
+    """
+
+    def __init__(self, max_workers: int) -> None:
+        self._executor = ThreadPoolExecutor(max_workers=max_workers)
+
+    def submit(self, fn: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]:
+        return self._executor.submit(fn, *args, **kwargs)
+
+    def shutdown(self, wait: bool = True) -> None:
+        self._executor.shutdown(wait=wait)
+
+
+class S3AioExecutor(S3Executor):
+    """Executor that schedules work on an asyncio event loop.
+
+    Uses ``asyncio.run_coroutine_threadsafe(asyncio.to_thread(fn), loop)`` to
+    dispatch blocking functions onto the event loop's thread pool, returning
+    ``concurrent.futures.Future`` objects that are compatible with
+    ``as_completed()`` and ``Future.cancel()``.
+
+    This avoids thread-in-thread nesting when ``S3File`` is used from within
+    ``asyncio.to_thread()`` calls (the pattern used by ``AioS3FileSystem``).
+
+    Args:
+        loop: A running asyncio event loop.
+
+    Raises:
+        RuntimeError: If the event loop is not running when ``submit`` is called.
+    """
+
+    def __init__(self, loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
+        self._loop = loop
+
+    def submit(self, fn: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]:
+        if self._loop is not None and self._loop.is_running():
+            return asyncio.run_coroutine_threadsafe(
+                asyncio.to_thread(fn, *args, **kwargs), self._loop
+            )
+        raise RuntimeError(
+            "S3AioExecutor requires a running event loop. "
+            "Use S3ThreadPoolExecutor for synchronous usage."
+        )
+
+    def shutdown(self, wait: bool = True) -> None:
+        # No resources to release — work is dispatched to the event loop.
+        pass