From 2169c92d63aa8316f5d060c9504dc9086cd21048 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Tue, 3 Nov 2026 21:29:06 +0900
Subject: [PATCH 1/4] Make files usable after fork with either filesystem
 (GH-1336)

S3ThreadPoolExecutor runs its calls in an S3AioThreadPool, which creates
new threads in a forked child process, so that work submitted there after
the executor ran a call in the parent no longer waits forever.

AioS3FileSystem._create_executor() reads the event loop through fsspec's
loop property, which raises RuntimeError after a fork, and S3AioExecutor
raises from submit() and only shuts down its pool in a forked child,
instead of scheduling onto, or waiting for, the parent's loop, whose copy
reports running but never runs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/aio.md                                   |  8 +-
 pyathena/filesystem/s3_async.py               |  7 +-
 pyathena/filesystem/s3_executor.py            | 33 ++++++--
 tests/pyathena/filesystem/test_s3_async.py    | 19 +++++
 tests/pyathena/filesystem/test_s3_executor.py | 79 +++++++++++++------
 tests/pyathena/util.py                        | 37 +++++++++
 6 files changed, 150 insertions(+), 33 deletions(-)

diff --git a/docs/aio.md b/docs/aio.md
index 1a6016068..29b5b51c9 100644
--- a/docs/aio.md
+++ b/docs/aio.md
@@ -334,7 +334,8 @@ multipart uploads.
 `S3FileSystem` and `S3File` use a pluggable executor abstraction (`S3Executor`) for
 parallel operations. Two implementations are provided:
 
-- `S3ThreadPoolExecutor` — wraps `ThreadPoolExecutor` (default for sync usage)
+- `S3ThreadPoolExecutor` — runs the calls in a pool of `max_workers` threads (default
+  for sync usage); in a forked child process, the pool creates new threads
 - `S3AioExecutor` — dispatches work via `asyncio.run_coroutine_threadsafe` to an
   `S3AioThreadPool` of its own
 
@@ -343,6 +344,11 @@ so multipart uploads and parallel range reads are dispatched through the event l
 a pool of `max_workers` threads of each file.
 An instance created with `asynchronous=True` has no event loop of its own, so its file handles use `S3ThreadPoolExecutor`.
 
+In a forked child process, the event loop of the parent does not run.
+An `AioS3FileSystem` created in the parent raises `RuntimeError` from `open()` there, as
+fsspec's synchronous methods do, and so does `S3AioExecutor.submit()`; a filesystem
+created in the child works.
+
 ### Async file handles
 
 `open_async()` returns an `AioS3StreamedFile`, whose `read()`, `write()`, and `close()`
diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py
index 4d93376f6..dfbcd840d 100644
--- a/pyathena/filesystem/s3_async.py
+++ b/pyathena/filesystem/s3_async.py
@@ -848,10 +848,15 @@ def _create_executor(self, max_workers: int) -> S3Executor:
         Returns:
             An ``S3AioExecutor`` on the event loop of this filesystem, or an
             ``S3ThreadPoolExecutor`` if it has none.
+
+        Raises:
+            RuntimeError: If called in a forked child process of the process
+                that created this instance, whose event loop does not run
+                there.
         """
         if self._loop is None:
             return S3ThreadPoolExecutor(max_workers=max_workers)
-        return S3AioExecutor(loop=self._loop, max_workers=max_workers)
+        return S3AioExecutor(loop=self.loop, max_workers=max_workers)
 
     def _open(
         self,
diff --git a/pyathena/filesystem/s3_executor.py b/pyathena/filesystem/s3_executor.py
index 9b22cdbc0..429808bbb 100644
--- a/pyathena/filesystem/s3_executor.py
+++ b/pyathena/filesystem/s3_executor.py
@@ -178,28 +178,32 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
 
 
 class S3ThreadPoolExecutor(S3Executor):
-    """Executor that delegates to a ``ThreadPoolExecutor``.
+    """Executor that runs the submitted calls in a pool of threads.
 
     This is the default executor used by ``S3File`` and ``S3FileSystem``
-    for synchronous parallel operations.
+    for synchronous parallel operations. The calls run in an
+    :class:`S3AioThreadPool`, so that a forked child process, whose copy of
+    the executor has no threads, creates new ones.
     """
 
     def __init__(self, max_workers: int) -> None:
-        """Initialize the executor with a new ``ThreadPoolExecutor``.
+        """Initialize the executor without starting any thread.
 
         Args:
             max_workers: The maximum number of threads of the thread pool.
+
+        Raises:
+            ValueError: If ``max_workers`` is not positive.
         """
-        self._executor = ThreadPoolExecutor(max_workers=max_workers)
+        self._pool = S3AioThreadPool(max_workers)
 
     @override
     def submit(self, fn: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]:
-        context = contextvars.copy_context()
-        return self._executor.submit(context.run, fn, *args, **kwargs)
+        return self._pool.submit(fn, *args, **kwargs)
 
     @override
     def shutdown(self, wait: bool = True) -> None:
-        self._executor.shutdown(wait=wait)
+        self._pool.shutdown(wait=wait)
 
 
 class S3AioThreadPool(S3Executor):
@@ -498,7 +502,9 @@ class S3AioExecutor(S3Executor):
         max_workers: The maximum number of submitted functions that run at once.
 
     Raises:
-        RuntimeError: If the event loop is not running when ``submit`` is called.
+        RuntimeError: If the event loop is not running when ``submit`` is
+            called, or ``submit`` is called in a forked child process, where
+            the event loop of the parent does not run.
     """
 
     def __init__(
@@ -518,6 +524,7 @@ def __init__(
             ValueError: If ``max_workers`` is not positive.
         """
         self._loop = loop
+        self._pid = os.getpid()
         self._pool = S3AioThreadPool(max_workers)
         # Orders the dispatches with the shutdown.
         self._lock = threading.Lock()
@@ -525,6 +532,9 @@ def __init__(
 
     @override
     def submit(self, fn: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]:
+        if self._pid != os.getpid():
+            # The copy of the loop reports running, but nothing runs it.
+            raise RuntimeError("S3AioExecutor cannot be used in a forked child process.")
         if self._loop is not None and self._loop.is_running():
             # The future of run_coroutine_threadsafe can be cancelled while
             # the function keeps running in its thread, so the returned future
@@ -568,6 +578,13 @@ async def _shutdown_pool(self) -> None:
 
     @override
     def shutdown(self, wait: bool = True) -> None:
+        if self._pid != os.getpid():
+            # In a forked child, no function has been dispatched to the
+            # pool, whose threads and lock have been reset, and the lock of
+            # this executor may be held by a thread of the parent.
+            self._shutdown = True
+            self._pool.shutdown(wait=wait)
+            return
         loop = self._loop
         with self._lock:
             self._shutdown = True
diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py
index b04dd7e32..95f2a1db7 100644
--- a/tests/pyathena/filesystem/test_s3_async.py
+++ b/tests/pyathena/filesystem/test_s3_async.py
@@ -47,6 +47,7 @@
     overlapping_calls,
     retried,
     run_bounded,
+    run_in_forked_child,
     shutdown_bounded,
     stub_multipart_copy,
     wait_until,
@@ -127,6 +128,24 @@ async def _multipart_copy(release, fs=None, **kwargs):
                 await asyncio.to_thread(shutdown_bounded, fs._thread_pool)
                 stubber.assert_no_pending_responses()
 
+    @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork() is not available.")
+    @pytest.mark.filterwarnings("ignore:.*fork\\(\\) may lead to deadlocks:DeprecationWarning")
+    def test_open_in_forked_child(self):
+        # GH-1336: in a forked child, a filesystem created in the parent
+        # raises on open(), as fsspec's synchronous methods do there. It used
+        # to return a file whose reads waited forever for the event loop of
+        # the parent.
+        fs = AioS3FileSystem(
+            key="dummy", secret="dummy", region_name="us-east-1", skip_instance_cache=True
+        )
+
+        def child():
+            with pytest.raises(RuntimeError, match="not fork-safe"):
+                fs.open("s3://bucket/key", "rb")
+            return True
+
+        assert run_in_forked_child(child) == 0
+
     @pytest.mark.parametrize("size", [0, 10])
     @pytest.mark.asyncio
     async def test_copy_object_with_multipart_upload_small_head_object_size(self, size):
diff --git a/tests/pyathena/filesystem/test_s3_executor.py b/tests/pyathena/filesystem/test_s3_executor.py
index ef4de1767..760f38823 100644
--- a/tests/pyathena/filesystem/test_s3_executor.py
+++ b/tests/pyathena/filesystem/test_s3_executor.py
@@ -9,7 +9,6 @@
 import contextvars
 import gc
 import os
-import signal
 import threading
 import time
 import traceback
@@ -23,7 +22,12 @@
 
 from pyathena.filesystem.s3_executor import S3AioExecutor, S3AioThreadPool, S3ThreadPoolExecutor
 from pyathena.util import RetryConfig, retry_api_call, stop_retries_on
-from tests.pyathena.util import EVENT_TIMEOUT, ThrottledOperation, overlapping_calls
+from tests.pyathena.util import (
+    EVENT_TIMEOUT,
+    ThrottledOperation,
+    overlapping_calls,
+    run_in_forked_child,
+)
 
 
 class TestS3AioThreadPool:
@@ -295,32 +299,16 @@ def hold_lock():
         holder.start()
         try:
             assert held.wait(EVENT_TIMEOUT)
-            pid = os.fork()
-        finally:
-            release.set()
-        if pid == 0:
-            # The child exits here, whatever happens, instead of running pytest.
-            code = 1
-            try:
-                if pool.submit(sum, [3, 4]).result(timeout=EVENT_TIMEOUT) == 7:
-                    code = 0
-            finally:
-                os._exit(code)
-        status = (0, 0)
-        try:
             # A child whose pool waits for the threads of the parent would
             # not exit.
-            deadline = time.monotonic() + 2 * EVENT_TIMEOUT
-            while (status := os.waitpid(pid, os.WNOHANG)) == (0, 0):
-                assert time.monotonic() < deadline, "the child did not exit"
-                time.sleep(0.05)
+            code = run_in_forked_child(
+                lambda: pool.submit(sum, [3, 4]).result(timeout=EVENT_TIMEOUT) == 7
+            )
         finally:
-            if status == (0, 0):
-                os.kill(pid, signal.SIGKILL)
-                os.waitpid(pid, 0)
+            release.set()
             holder.join(EVENT_TIMEOUT)
             pool.shutdown()
-        assert os.waitstatus_to_exitcode(status[1]) == 0
+        assert code == 0
 
 
 class TestS3AioExecutor:
@@ -502,6 +490,33 @@ def block():
             loop.run_until_complete(asyncio.sleep(0))
             loop.close()
 
+    @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork() is not available.")
+    @pytest.mark.filterwarnings("ignore:.*fork\\(\\) may lead to deadlocks:DeprecationWarning")
+    def test_fork(self):
+        # GH-1336: in a forked child, where the event loop of the parent does
+        # not run, submit() raises and shutdown() returns. Both used to wait
+        # forever for the copy of the loop, which reports running.
+        loop = asyncio.new_event_loop()
+        thread = threading.Thread(target=loop.run_forever)
+        thread.start()
+        executor = S3AioExecutor(loop=loop)
+
+        def child():
+            with pytest.raises(RuntimeError, match="forked child"):
+                executor.submit(sum, [3, 4])
+            executor.shutdown(wait=True)
+            return True
+
+        try:
+            assert executor.submit(sum, [1, 2]).result(timeout=EVENT_TIMEOUT) == 3
+            code = run_in_forked_child(child)
+        finally:
+            executor.shutdown()
+            loop.call_soon_threadsafe(loop.stop)
+            thread.join(EVENT_TIMEOUT)
+            loop.close()
+        assert code == 0
+
     def test_submit_without_running_loop(self):
         with pytest.raises(RuntimeError, match="requires a running event loop"):
             S3AioExecutor().submit(sum, [1, 2])
@@ -545,6 +560,24 @@ async def main():
         assert pending.cancelled()
 
 
+class TestS3ThreadPoolExecutor:
+    @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork() is not available.")
+    @pytest.mark.filterwarnings("ignore:.*fork\\(\\) may lead to deadlocks:DeprecationWarning")
+    def test_fork(self):
+        # GH-1336: a forked child, whose copy of the executor has no threads,
+        # runs the calls submitted there. After the executor had run a call
+        # in the parent, they used to wait forever.
+        executor = S3ThreadPoolExecutor(max_workers=1)
+        try:
+            assert executor.submit(sum, [1, 2]).result(timeout=EVENT_TIMEOUT) == 3
+            code = run_in_forked_child(
+                lambda: executor.submit(sum, [3, 4]).result(timeout=EVENT_TIMEOUT) == 7
+            )
+        finally:
+            executor.shutdown()
+        assert code == 0
+
+
 class TestS3Executor:
     @pytest.mark.parametrize("executor_class", [S3ThreadPoolExecutor, S3AioThreadPool])
     def test_submit_context(self, executor_class):
diff --git a/tests/pyathena/util.py b/tests/pyathena/util.py
index 89ac327ce..24d94a2c9 100644
--- a/tests/pyathena/util.py
+++ b/tests/pyathena/util.py
@@ -7,6 +7,8 @@
 
 import asyncio
 import io
+import os
+import signal
 import threading
 import time
 from concurrent.futures import Future, wait
@@ -277,6 +279,41 @@ def run():
     return outcome.get("result")
 
 
+def run_in_forked_child(child, timeout=2 * EVENT_TIMEOUT):
+    """Run a check in a forked child process, failing if the child does not exit in time.
+
+    The child exits with the status 0 if the check returns a true value, and
+    1 otherwise or if it raises, instead of returning to pytest. A child that
+    a regression blocks is killed, so that the test fails instead of hanging.
+
+    Args:
+        child: The check, called without arguments in the child.
+        timeout: The longest run of the child, in seconds.
+
+    Returns:
+        The exit code of the child.
+    """
+    pid = os.fork()
+    if pid == 0:
+        code = 1
+        try:
+            if child():
+                code = 0
+        finally:
+            os._exit(code)
+    status = (0, 0)
+    try:
+        deadline = time.monotonic() + timeout
+        while (status := os.waitpid(pid, os.WNOHANG)) == (0, 0):
+            assert time.monotonic() < deadline, f"the child did not exit within {timeout} s"
+            time.sleep(0.05)
+    finally:
+        if status == (0, 0):
+            os.kill(pid, signal.SIGKILL)
+            os.waitpid(pid, 0)
+    return os.waitstatus_to_exitcode(status[1])
+
+
 def shutdown_bounded(pool, timeout=EVENT_TIMEOUT):
     """Shut down an executor and wait for its calls, failing if they run past the timeout.
 

From f192c0fedfa38b37be0ef3c56b2278d084b78922 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Tue, 3 Nov 2026 22:06:06 +0900
Subject: [PATCH 2/4] Reset S3AioExecutor in the at-fork hook and narrow the
 docs (GH-1336)

S3AioExecutor detaches from the parent's event loop and replaces its lock
in a register_at_fork hook, as S3AioThreadPool resets itself, instead of
comparing process IDs in submit() and shutdown(); shutdown() then takes
its existing path for an executor without a loop.

The docs and the docstring of _create_executor() state that only an
AioS3FileSystem with an event loop of its own raises after a fork. Fork
tests share a forking decorator, the child prints the traceback of a
failed check, and the S3AioExecutor fork test holds the executor's lock
across the fork.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/aio.md                                   |  6 ++--
 pyathena/filesystem/s3_async.py               |  6 ++--
 pyathena/filesystem/s3_executor.py            | 35 +++++++++++++------
 tests/pyathena/filesystem/test_s3_async.py    |  4 +--
 tests/pyathena/filesystem/test_s3_executor.py | 26 +++++++++-----
 tests/pyathena/util.py                        | 26 +++++++++++++-
 6 files changed, 76 insertions(+), 27 deletions(-)

diff --git a/docs/aio.md b/docs/aio.md
index 29b5b51c9..ab83aea8a 100644
--- a/docs/aio.md
+++ b/docs/aio.md
@@ -345,9 +345,9 @@ a pool of `max_workers` threads of each file.
 An instance created with `asynchronous=True` has no event loop of its own, so its file handles use `S3ThreadPoolExecutor`.
 
 In a forked child process, the event loop of the parent does not run.
-An `AioS3FileSystem` created in the parent raises `RuntimeError` from `open()` there, as
-fsspec's synchronous methods do, and so does `S3AioExecutor.submit()`; a filesystem
-created in the child works.
+An `AioS3FileSystem` created in the parent without `asynchronous=True` raises
+`RuntimeError` from `open()` there, as fsspec's synchronous methods do, and so does
+`S3AioExecutor.submit()`; a filesystem created in the child works.
 
 ### Async file handles
 
diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py
index dfbcd840d..57203a7d0 100644
--- a/pyathena/filesystem/s3_async.py
+++ b/pyathena/filesystem/s3_async.py
@@ -850,9 +850,9 @@ def _create_executor(self, max_workers: int) -> S3Executor:
             ``S3ThreadPoolExecutor`` if it has none.
 
         Raises:
-            RuntimeError: If called in a forked child process of the process
-                that created this instance, whose event loop does not run
-                there.
+            RuntimeError: If this instance has an event loop and the call is
+                made in a forked child process of the process that created
+                it, where the loop does not run.
         """
         if self._loop is None:
             return S3ThreadPoolExecutor(max_workers=max_workers)
diff --git a/pyathena/filesystem/s3_executor.py b/pyathena/filesystem/s3_executor.py
index 429808bbb..1e9b3f2e6 100644
--- a/pyathena/filesystem/s3_executor.py
+++ b/pyathena/filesystem/s3_executor.py
@@ -507,6 +507,10 @@ class S3AioExecutor(S3Executor):
             the event loop of the parent does not run.
     """
 
+    # The executors of this process, which a forked child detaches from the
+    # event loop of its parent.
+    _instances: ClassVar[weakref.WeakSet[S3AioExecutor]] = weakref.WeakSet()
+
     def __init__(
         self,
         loop: asyncio.AbstractEventLoop | None = None,
@@ -524,16 +528,16 @@ def __init__(
             ValueError: If ``max_workers`` is not positive.
         """
         self._loop = loop
-        self._pid = os.getpid()
         self._pool = S3AioThreadPool(max_workers)
         # Orders the dispatches with the shutdown.
         self._lock = threading.Lock()
         self._shutdown = False
+        self._forked = False
+        S3AioExecutor._instances.add(self)
 
     @override
     def submit(self, fn: Callable[..., T], *args: Any, **kwargs: Any) -> Future[T]:
-        if self._pid != os.getpid():
-            # The copy of the loop reports running, but nothing runs it.
+        if self._forked:
             raise RuntimeError("S3AioExecutor cannot be used in a forked child process.")
         if self._loop is not None and self._loop.is_running():
             # The future of run_coroutine_threadsafe can be cancelled while
@@ -578,13 +582,6 @@ async def _shutdown_pool(self) -> None:
 
     @override
     def shutdown(self, wait: bool = True) -> None:
-        if self._pid != os.getpid():
-            # In a forked child, no function has been dispatched to the
-            # pool, whose threads and lock have been reset, and the lock of
-            # this executor may be held by a thread of the parent.
-            self._shutdown = True
-            self._pool.shutdown(wait=wait)
-            return
         loop = self._loop
         with self._lock:
             self._shutdown = True
@@ -613,3 +610,21 @@ def shutdown(self, wait: bool = True) -> None:
                 if not loop.is_running() or sys.is_finalizing():
                     break
             self._pool.shutdown(wait=True)
+
+    @classmethod
+    def _reset_after_fork(cls) -> None:
+        """Detach the executors that a forked child copied from the event loop of its parent.
+
+        The copy of the loop reports running, but nothing runs it, and a lock
+        that a thread of the parent held stays held. Each executor then
+        raises from :meth:`submit`, and :meth:`shutdown` only shuts down its
+        pool.
+        """
+        for executor in list(cls._instances):
+            executor._loop = None
+            executor._lock = threading.Lock()
+            executor._forked = True
+
+
+if hasattr(os, "register_at_fork"):
+    os.register_at_fork(after_in_child=S3AioExecutor._reset_after_fork)
diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py
index 95f2a1db7..ff98fd09b 100644
--- a/tests/pyathena/filesystem/test_s3_async.py
+++ b/tests/pyathena/filesystem/test_s3_async.py
@@ -44,6 +44,7 @@
     ThrottledOperation,
     await_bounded,
     cancel_when_started,
+    forking,
     overlapping_calls,
     retried,
     run_bounded,
@@ -128,8 +129,7 @@ async def _multipart_copy(release, fs=None, **kwargs):
                 await asyncio.to_thread(shutdown_bounded, fs._thread_pool)
                 stubber.assert_no_pending_responses()
 
-    @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork() is not available.")
-    @pytest.mark.filterwarnings("ignore:.*fork\\(\\) may lead to deadlocks:DeprecationWarning")
+    @forking
     def test_open_in_forked_child(self):
         # GH-1336: in a forked child, a filesystem created in the parent
         # raises on open(), as fsspec's synchronous methods do there. It used
diff --git a/tests/pyathena/filesystem/test_s3_executor.py b/tests/pyathena/filesystem/test_s3_executor.py
index 760f38823..9f7e9e930 100644
--- a/tests/pyathena/filesystem/test_s3_executor.py
+++ b/tests/pyathena/filesystem/test_s3_executor.py
@@ -8,7 +8,6 @@
 import asyncio
 import contextvars
 import gc
-import os
 import threading
 import time
 import traceback
@@ -25,6 +24,7 @@
 from tests.pyathena.util import (
     EVENT_TIMEOUT,
     ThrottledOperation,
+    forking,
     overlapping_calls,
     run_in_forked_child,
 )
@@ -279,8 +279,7 @@ def test_release_after_interrupted_submit(self):
         assert not threads[0].is_alive()
         pool.shutdown()
 
-    @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork() is not available.")
-    @pytest.mark.filterwarnings("ignore:.*fork\\(\\) may lead to deadlocks:DeprecationWarning")
+    @forking
     def test_fork(self):
         # A forked child, whose copy of the pool has no threads, creates new
         # ones, also if a thread of the parent held the lock of the pool.
@@ -490,16 +489,22 @@ def block():
             loop.run_until_complete(asyncio.sleep(0))
             loop.close()
 
-    @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork() is not available.")
-    @pytest.mark.filterwarnings("ignore:.*fork\\(\\) may lead to deadlocks:DeprecationWarning")
+    @forking
     def test_fork(self):
         # GH-1336: in a forked child, where the event loop of the parent does
-        # not run, submit() raises and shutdown() returns. Both used to wait
+        # not run, submit() raises and shutdown() returns, also if a thread
+        # of the parent held the lock of the executor. Both used to wait
         # forever for the copy of the loop, which reports running.
         loop = asyncio.new_event_loop()
         thread = threading.Thread(target=loop.run_forever)
         thread.start()
         executor = S3AioExecutor(loop=loop)
+        held, release = threading.Event(), threading.Event()
+
+        def hold_lock():
+            with executor._lock:
+                held.set()
+                release.wait(EVENT_TIMEOUT)
 
         def child():
             with pytest.raises(RuntimeError, match="forked child"):
@@ -507,10 +512,16 @@ def child():
             executor.shutdown(wait=True)
             return True
 
+        holder = threading.Thread(target=hold_lock)
         try:
             assert executor.submit(sum, [1, 2]).result(timeout=EVENT_TIMEOUT) == 3
+            holder.start()
+            assert held.wait(EVENT_TIMEOUT)
             code = run_in_forked_child(child)
         finally:
+            release.set()
+            if holder.ident is not None:
+                holder.join(EVENT_TIMEOUT)
             executor.shutdown()
             loop.call_soon_threadsafe(loop.stop)
             thread.join(EVENT_TIMEOUT)
@@ -561,8 +572,7 @@ async def main():
 
 
 class TestS3ThreadPoolExecutor:
-    @pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork() is not available.")
-    @pytest.mark.filterwarnings("ignore:.*fork\\(\\) may lead to deadlocks:DeprecationWarning")
+    @forking
     def test_fork(self):
         # GH-1336: a forked child, whose copy of the executor has no threads,
         # runs the calls submitted there. After the executor had run a call
diff --git a/tests/pyathena/util.py b/tests/pyathena/util.py
index 24d94a2c9..277b020fe 100644
--- a/tests/pyathena/util.py
+++ b/tests/pyathena/util.py
@@ -11,6 +11,7 @@
 import signal
 import threading
 import time
+import traceback
 from concurrent.futures import Future, wait
 from contextlib import contextmanager
 from datetime import UTC, datetime, timedelta, timezone
@@ -279,11 +280,31 @@ def run():
     return outcome.get("result")
 
 
+def forking(test):
+    """Mark a test that forks a child process.
+
+    The test is skipped where ``os.fork()`` is not available, and the warning
+    of Python 3.12 and later about forking a multi-threaded process is
+    ignored.
+
+    Args:
+        test: The test function.
+
+    Returns:
+        The marked test function.
+    """
+    test = pytest.mark.filterwarnings(
+        "ignore:.*fork\\(\\) may lead to deadlocks:DeprecationWarning"
+    )(test)
+    return pytest.mark.skipif(not hasattr(os, "fork"), reason="os.fork() is not available.")(test)
+
+
 def run_in_forked_child(child, timeout=2 * EVENT_TIMEOUT):
     """Run a check in a forked child process, failing if the child does not exit in time.
 
     The child exits with the status 0 if the check returns a true value, and
-    1 otherwise or if it raises, instead of returning to pytest. A child that
+    1 otherwise or if it raises, after printing the traceback, instead of
+    returning to pytest. A child that
     a regression blocks is killed, so that the test fails instead of hanging.
 
     Args:
@@ -299,6 +320,9 @@ def run_in_forked_child(child, timeout=2 * EVENT_TIMEOUT):
         try:
             if child():
                 code = 0
+        except BaseException:
+            # Shown in the captured output of the failed test.
+            traceback.print_exc()
         finally:
             os._exit(code)
     status = (0, 0)

From 7fafe8d2260b7fb46ec70ecc500d727370a81c42 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Tue, 3 Nov 2026 23:11:46 +0900
Subject: [PATCH 3/4] Wait for the event loop to run in the S3AioExecutor fork
 test (GH-1336)

S3AioExecutor submits only to a running loop, and the thread that runs
it may not have started the loop yet when the test first submits.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/pyathena/filesystem/test_s3_executor.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/tests/pyathena/filesystem/test_s3_executor.py b/tests/pyathena/filesystem/test_s3_executor.py
index 9f7e9e930..dfe8b7881 100644
--- a/tests/pyathena/filesystem/test_s3_executor.py
+++ b/tests/pyathena/filesystem/test_s3_executor.py
@@ -513,7 +513,11 @@ def child():
             return True
 
         holder = threading.Thread(target=hold_lock)
+        # The executor submits only to a loop that runs.
+        running = threading.Event()
+        loop.call_soon_threadsafe(running.set)
         try:
+            assert running.wait(EVENT_TIMEOUT)
             assert executor.submit(sum, [1, 2]).result(timeout=EVENT_TIMEOUT) == 3
             holder.start()
             assert held.wait(EVENT_TIMEOUT)

From e2d2496bbf8e7f0cc2b635afe8f1989253ead1b3 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Tue, 3 Nov 2026 23:18:52 +0900
Subject: [PATCH 4/4] Bound the cleanup of the executor fork tests (GH-1336)

The lock holders keep the lock until the test releases it, so that a
slow parent cannot fork with the lock already free, and run as daemon
threads. The S3AioExecutor fork test shuts the executor down with a
bounded wait, runs its loop in a daemon thread, checks that the loop
thread ended, and runs the pending dispatch tasks before closing it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/pyathena/filesystem/test_s3_executor.py | 16 ++++++++++------
 1 file changed, 10 insertions(+), 6 deletions(-)

diff --git a/tests/pyathena/filesystem/test_s3_executor.py b/tests/pyathena/filesystem/test_s3_executor.py
index dfe8b7881..d6a7bdc51 100644
--- a/tests/pyathena/filesystem/test_s3_executor.py
+++ b/tests/pyathena/filesystem/test_s3_executor.py
@@ -27,6 +27,7 @@
     forking,
     overlapping_calls,
     run_in_forked_child,
+    shutdown_bounded,
 )
 
 
@@ -292,9 +293,9 @@ def test_fork(self):
         def hold_lock():
             with pool._lock:
                 held.set()
-                release.wait(EVENT_TIMEOUT)
+                release.wait()
 
-        holder = threading.Thread(target=hold_lock)
+        holder = threading.Thread(target=hold_lock, daemon=True)
         holder.start()
         try:
             assert held.wait(EVENT_TIMEOUT)
@@ -496,7 +497,7 @@ def test_fork(self):
         # of the parent held the lock of the executor. Both used to wait
         # forever for the copy of the loop, which reports running.
         loop = asyncio.new_event_loop()
-        thread = threading.Thread(target=loop.run_forever)
+        thread = threading.Thread(target=loop.run_forever, daemon=True)
         thread.start()
         executor = S3AioExecutor(loop=loop)
         held, release = threading.Event(), threading.Event()
@@ -504,7 +505,7 @@ def test_fork(self):
         def hold_lock():
             with executor._lock:
                 held.set()
-                release.wait(EVENT_TIMEOUT)
+                release.wait()
 
         def child():
             with pytest.raises(RuntimeError, match="forked child"):
@@ -512,7 +513,7 @@ def child():
             executor.shutdown(wait=True)
             return True
 
-        holder = threading.Thread(target=hold_lock)
+        holder = threading.Thread(target=hold_lock, daemon=True)
         # The executor submits only to a loop that runs.
         running = threading.Event()
         loop.call_soon_threadsafe(running.set)
@@ -526,9 +527,12 @@ def child():
             release.set()
             if holder.ident is not None:
                 holder.join(EVENT_TIMEOUT)
-            executor.shutdown()
+            shutdown_bounded(executor)
             loop.call_soon_threadsafe(loop.stop)
             thread.join(EVENT_TIMEOUT)
+            assert not thread.is_alive()
+            # Runs the dispatch tasks that the stop left pending.
+            loop.run_until_complete(asyncio.sleep(0))
             loop.close()
         assert code == 0