From f7e869070fdf674979e12e24399bcd1e29799885 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 12:08:29 +0900
Subject: [PATCH 1/9] Invalidate tuple-keyed listings in
 S3FileSystem.invalidate_cache

_ls_dirs() caches listings under (path, delimiter) since v3.15.0, but
invalidate_cache(path) only popped string keys. rm(), touch(),
pipe_file(), S3File.commit() and the other writes rely on it, so ls()
and find() kept returning stale listings. Drop the "/" and "" listing
entries of the path and its parents as well.

The cache key also ignored prefix and next_token, so find(path,
prefix=...) could return the unfiltered listing of path. Skip the cache
for these partial listings.

Closes #922

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py            | 28 ++++++++++--
 tests/pyathena/filesystem/test_s3.py | 68 ++++++++++++++++++++++++++++
 2 files changed, 93 insertions(+), 3 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 4104f9354..e5aeaf73b 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -404,13 +404,32 @@ def _ls_dirs(
         max_keys: int | None = None,
         refresh: bool = False,
     ) -> list[S3Object]:
+        """List the objects and common prefixes under a path.
+
+        A complete listing of the path is cached under ``(path, delimiter)``,
+        which ``invalidate_cache`` drops for the path and its parents.
+
+        Args:
+            path: The bucket or directory path to list.
+            prefix: Key prefix to filter by, relative to the path. A prefixed
+                listing is neither read from nor written to the cache.
+            delimiter: Delimiter to group keys by; ``""`` lists recursively.
+            next_token: Continuation token to start listing from. A listing
+                that starts from a token is neither read from nor written to
+                the cache.
+            max_keys: Maximum number of keys per ListObjectsV2 request.
+            refresh: If True, bypass the cache and list from S3.
+
+        Returns:
+            The listed directories and files.
+        """
         bucket, key, version_id = self.parse_path(path)
+        use_cache = not prefix and not next_token
         if key:
             prefix = f"{key}/{prefix if prefix else ''}"
 
-        # Create a cache key that includes the delimiter
         cache_key = (path, delimiter)
-        if cache_key in self.dircache and not refresh:
+        if use_cache and cache_key in self.dircache and not refresh:
             return cast(list[S3Object], self.dircache[cache_key])
 
         files: list[S3Object] = []
@@ -444,7 +463,7 @@ def _ls_dirs(
             next_token = response.get("NextContinuationToken")
             if not next_token:
                 break
-        if files:
+        if use_cache and files:
             self.dircache[cache_key] = files
         return files
 
@@ -1882,6 +1901,9 @@ def invalidate_cache(self, path: str | None = None) -> None:
             path = self._strip_protocol(path)
             while path:
                 self.dircache.pop(path, None)
+                # _ls_dirs caches listings under (path, delimiter).
+                for delimiter in ("/", ""):
+                    self.dircache.pop((path, delimiter), None)
                 path = self._parent(path)
 
     def _ls_from_cache(self, path: str) -> list[S3Object] | S3Object | None:
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index a2ead0af2..f6f6c1f76 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -192,6 +192,51 @@ def test_ls_from_cache_with_cached_object(self):
         # every cache value is a listing and raises TypeError here).
         assert fs._ls_from_cache("bucket/key/child") is None
 
+    def test_invalidate_cache_drops_listings_of_path_and_parents(self):
+        fs = self._make_fs()
+        invalidated = [
+            "bucket/a/b/c.txt",
+            ("bucket/a/b", "/"),
+            ("bucket/a/b", ""),
+            ("bucket/a", "/"),
+            ("bucket/a", ""),
+            ("bucket", "/"),
+            ("bucket", ""),
+        ]
+        kept = ["", ("bucket/a/x", "/"), ("bucket/a/b/c.txt/d", "")]
+        for cache_key in invalidated + kept:
+            fs.dircache[cache_key] = []
+
+        fs.invalidate_cache("s3://bucket/a/b/c.txt")
+        assert list(fs.dircache) == kept
+
+    @pytest.mark.parametrize(
+        ("prefix", "next_token"),
+        [
+            ("test_", None),
+            ("", "token"),
+        ],
+    )
+    def test_ls_dirs_partial_listing_bypasses_cache(self, prefix, next_token):
+        fs = self._make_fs()
+        cached = S3Object(
+            init={"Key": "dir/cached"},
+            type=S3ObjectType.S3_OBJECT_TYPE_FILE,
+            bucket="bucket",
+            key="dir/cached",
+        )
+        fs.dircache[("bucket/dir", "")] = [cached]
+        fs._call.return_value = {"Contents": [{"Key": "dir/test_1"}]}
+
+        files = fs._ls_dirs("bucket/dir", prefix=prefix, delimiter="", next_token=next_token)
+        assert [f.name for f in files] == ["bucket/dir/test_1"]
+        assert fs.dircache[("bucket/dir", "")] == [cached]
+
+        # A complete listing of the path is still served from the cache.
+        fs._call.reset_mock()
+        assert fs._ls_dirs("bucket/dir", delimiter="") == [cached]
+        fs._call.assert_not_called()
+
     def test_mkdir_creates_bucket(self):
         fs = self._make_fs()
         fs.allow_bucket_creation = True
@@ -719,6 +764,29 @@ def test_ls_dirs(self, fs):
         assert test_1_detail[0].name == fs._strip_protocol(f"{dir_}/prefix/test_1")
         assert test_1_detail[0].size == 1
 
+    def test_ls_and_find_reflect_changes_through_the_filesystem(self, fs):
+        dir_ = (
+            f"s3://{ENV.s3_staging_bucket}/{ENV.s3_staging_key}{ENV.schema}/"
+            f"filesystem/test_ls_and_find_reflect_changes/{uuid.uuid4()}"
+        )
+        path = fs._strip_protocol(dir_)
+        fs.touch(f"{dir_}/a.txt")
+        fs.touch(f"{dir_}/b.txt")
+        assert sorted(fs.ls(dir_)) == [f"{path}/a.txt", f"{path}/b.txt"]
+        assert sorted(fs.find(dir_)) == [f"{path}/a.txt", f"{path}/b.txt"]
+
+        fs.rm(f"{dir_}/a.txt")
+        assert fs.ls(dir_) == [f"{path}/b.txt"]
+        assert fs.find(dir_) == [f"{path}/b.txt"]
+
+        fs.touch(f"{dir_}/c.txt")
+        assert sorted(fs.ls(dir_)) == [f"{path}/b.txt", f"{path}/c.txt"]
+        assert sorted(fs.find(dir_)) == [f"{path}/b.txt", f"{path}/c.txt"]
+        # A prefixed find must not be served from the unprefixed listing.
+        assert fs.find(dir_, prefix="c") == [f"{path}/c.txt"]
+
+        fs.rm(dir_, recursive=True)
+
     def test_info_bucket(self, fs):
         dir_ = f"s3://{ENV.s3_staging_bucket}"
         bucket, key, version_id = fs.parse_path(dir_)

From 4d218948871836f81bfe74be231c81de4fe124c2 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 12:10:57 +0900
Subject: [PATCH 2/9] Do not pin descendant listings in the invalidate_cache
 test

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

diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index f6f6c1f76..b0c70f069 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -203,7 +203,7 @@ def test_invalidate_cache_drops_listings_of_path_and_parents(self):
             ("bucket", "/"),
             ("bucket", ""),
         ]
-        kept = ["", ("bucket/a/x", "/"), ("bucket/a/b/c.txt/d", "")]
+        kept = ["", ("bucket/a/x", "/")]
         for cache_key in invalidated + kept:
             fs.dircache[cache_key] = []
 

From ab9f48a867645d771903877ca9cd84d57798b292 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 12:12:16 +0900
Subject: [PATCH 3/9] Clarify when the _ls_dirs listing cache is invalidated

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index e5aeaf73b..f2c191307 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -406,8 +406,9 @@ def _ls_dirs(
     ) -> list[S3Object]:
         """List the objects and common prefixes under a path.
 
-        A complete listing of the path is cached under ``(path, delimiter)``,
-        which ``invalidate_cache`` drops for the path and its parents.
+        A complete listing of the path is cached under ``(path, delimiter)``.
+        ``invalidate_cache`` drops it when the path or a path under it is
+        invalidated.
 
         Args:
             path: The bucket or directory path to list.

From 38a2e285d42a2b4f43d55858515e022501852269 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 12:28:09 +0900
Subject: [PATCH 4/9] Keep the cached listing intact on withdirs and empty
 refreshes

find(withdirs=True) extended the list returned by _ls_dirs() in place,
which is the cached listing, so each repeated call added the derived
directories again. Build a new list instead.

An empty refreshed listing neither replaced nor evicted the cached
entry, so the next ls() returned the deleted objects again. Evict it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py            | 13 +++++++----
 tests/pyathena/filesystem/test_s3.py | 35 ++++++++++++++++++++++++++++
 2 files changed, 44 insertions(+), 4 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index f2c191307..b44a184d6 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -406,7 +406,8 @@ def _ls_dirs(
     ) -> list[S3Object]:
         """List the objects and common prefixes under a path.
 
-        A complete listing of the path is cached under ``(path, delimiter)``.
+        A complete, non-empty listing of the path is cached under
+        ``(path, delimiter)``, and an empty one evicts it.
         ``invalidate_cache`` drops it when the path or a path under it is
         invalidated.
 
@@ -464,8 +465,11 @@ def _ls_dirs(
             next_token = response.get("NextContinuationToken")
             if not next_token:
                 break
-        if use_cache and files:
-            self.dircache[cache_key] = files
+        if use_cache:
+            if files:
+                self.dircache[cache_key] = files
+            else:
+                self.dircache.pop(cache_key, None)
         return files
 
     def ls(
@@ -753,7 +757,8 @@ def _find(
 
         # If withdirs is True, we need to derive directories from file paths
         if withdirs:
-            files.extend(self._extract_parent_directories(files, bucket, key))
+            # Build a new list; files may be the cached listing.
+            files = files + self._extract_parent_directories(files, bucket, key)
 
         # Filter directories if withdirs is False (default)
         if withdirs is False or withdirs is None:
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index b0c70f069..7d09e832d 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -237,6 +237,41 @@ def test_ls_dirs_partial_listing_bypasses_cache(self, prefix, next_token):
         assert fs._ls_dirs("bucket/dir", delimiter="") == [cached]
         fs._call.assert_not_called()
 
+    def test_ls_dirs_empty_refresh_evicts_cached_listing(self):
+        fs = self._make_fs()
+        fs.dircache[("bucket/dir", "/")] = [
+            S3Object(
+                init={"Key": "dir/deleted"},
+                type=S3ObjectType.S3_OBJECT_TYPE_FILE,
+                bucket="bucket",
+                key="dir/deleted",
+            )
+        ]
+        fs._call.return_value = {}
+
+        assert fs._ls_dirs("bucket/dir", refresh=True) == []
+        # The next listing must not return the deleted object from the cache.
+        fs._call.reset_mock()
+        assert fs._ls_dirs("bucket/dir") == []
+        fs._call.assert_called_once()
+
+    def test_find_withdirs_does_not_modify_cached_listing(self):
+        fs = self._make_fs()
+        fs.dircache[("bucket/dir", "")] = [
+            S3Object(
+                init={"Key": "dir/sub/file"},
+                type=S3ObjectType.S3_OBJECT_TYPE_FILE,
+                bucket="bucket",
+                key="dir/sub/file",
+            )
+        ]
+
+        expected = ["bucket/dir/sub", "bucket/dir/sub/file"]
+        assert sorted(fs.find("s3://bucket/dir", withdirs=True)) == expected
+        assert sorted(fs.find("s3://bucket/dir", withdirs=True)) == expected
+        assert fs.find("s3://bucket/dir") == ["bucket/dir/sub/file"]
+        fs._call.assert_not_called()
+
     def test_mkdir_creates_bucket(self):
         fs = self._make_fs()
         fs.allow_bucket_creation = True

From 4ad76746fc558eca62bea5d70419dbb1cfed29b2 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 12:39:20 +0900
Subject: [PATCH 5/9] Honor refresh in find() and evict objects that are no
 longer found

find(path, refresh=True) ignored refresh, so it kept returning the
cached listings. Forward it to _ls_dirs(), the info() fallback and the
recursive maxdepth calls.

A refreshed HeadObject or HeadBucket that found nothing left the cached
entry, so the next ls(), info() or exists() returned the deleted object
or bucket from the cache. Evict it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py            | 10 ++--
 tests/pyathena/filesystem/test_s3.py | 73 +++++++++++++++++++---------
 2 files changed, 58 insertions(+), 25 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index b44a184d6..2c3a3b279 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -315,6 +315,7 @@ def _head_bucket(self, bucket, refresh: bool = False) -> S3Object | None:
                     Bucket=bucket,
                 )
             except FileNotFoundError:
+                self.dircache.pop(bucket, None)
                 return None
             file = S3Object(
                 init={
@@ -352,6 +353,7 @@ def _head_object(
                     **request,
                 )
             except FileNotFoundError:
+                self.dircache.pop(path, None)
                 return None
             if self.version_aware and not version_id:
                 # Pin the version of the object so that subsequent reads see
@@ -720,13 +722,15 @@ def _find(
             raise ValueError("Cannot traverse all files in S3.")
         bucket, key, _ = self.parse_path(path)
         prefix = kwargs.pop("prefix", "")
+        # Keep refresh in kwargs so that the recursive calls also refresh.
+        refresh = kwargs.get("refresh", False)
 
         # When maxdepth is specified, use a recursive approach with delimiter
         if maxdepth is not None:
             result: list[S3Object] = []
 
             # List files and directories at current level
-            current_items = self._ls_dirs(path, prefix=prefix, delimiter="/")
+            current_items = self._ls_dirs(path, prefix=prefix, delimiter="/", refresh=refresh)
 
             for item in current_items:
                 if item.type == S3ObjectType.S3_OBJECT_TYPE_FILE:
@@ -748,10 +752,10 @@ def _find(
             return result
 
         # For unlimited depth, use the original approach (get all files at once)
-        files = self._ls_dirs(path, prefix=prefix, delimiter="")
+        files = self._ls_dirs(path, prefix=prefix, delimiter="", refresh=refresh)
         if not files and key:
             try:
-                files = [self.info(path)]
+                files = [self.info(path, refresh=refresh)]
             except FileNotFoundError:
                 files = []
 
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 7d09e832d..21ed555c6 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -148,6 +148,16 @@ def _make_fs():
         fs.version_aware = False
         return fs
 
+    @staticmethod
+    def _file_object(key):
+        # Build a listed file entry in the bucket named "bucket".
+        return S3Object(
+            init={"Key": key},
+            type=S3ObjectType.S3_OBJECT_TYPE_FILE,
+            bucket="bucket",
+            key=key,
+        )
+
     def test_get_client_compatible_with_s3fs(self):
         # Only constructs a boto3 client; no AWS access.
         fs = S3FileSystem(
@@ -219,12 +229,7 @@ def test_invalidate_cache_drops_listings_of_path_and_parents(self):
     )
     def test_ls_dirs_partial_listing_bypasses_cache(self, prefix, next_token):
         fs = self._make_fs()
-        cached = S3Object(
-            init={"Key": "dir/cached"},
-            type=S3ObjectType.S3_OBJECT_TYPE_FILE,
-            bucket="bucket",
-            key="dir/cached",
-        )
+        cached = self._file_object("dir/cached")
         fs.dircache[("bucket/dir", "")] = [cached]
         fs._call.return_value = {"Contents": [{"Key": "dir/test_1"}]}
 
@@ -239,14 +244,7 @@ def test_ls_dirs_partial_listing_bypasses_cache(self, prefix, next_token):
 
     def test_ls_dirs_empty_refresh_evicts_cached_listing(self):
         fs = self._make_fs()
-        fs.dircache[("bucket/dir", "/")] = [
-            S3Object(
-                init={"Key": "dir/deleted"},
-                type=S3ObjectType.S3_OBJECT_TYPE_FILE,
-                bucket="bucket",
-                key="dir/deleted",
-            )
-        ]
+        fs.dircache[("bucket/dir", "/")] = [self._file_object("dir/deleted")]
         fs._call.return_value = {}
 
         assert fs._ls_dirs("bucket/dir", refresh=True) == []
@@ -257,14 +255,7 @@ def test_ls_dirs_empty_refresh_evicts_cached_listing(self):
 
     def test_find_withdirs_does_not_modify_cached_listing(self):
         fs = self._make_fs()
-        fs.dircache[("bucket/dir", "")] = [
-            S3Object(
-                init={"Key": "dir/sub/file"},
-                type=S3ObjectType.S3_OBJECT_TYPE_FILE,
-                bucket="bucket",
-                key="dir/sub/file",
-            )
-        ]
+        fs.dircache[("bucket/dir", "")] = [self._file_object("dir/sub/file")]
 
         expected = ["bucket/dir/sub", "bucket/dir/sub/file"]
         assert sorted(fs.find("s3://bucket/dir", withdirs=True)) == expected
@@ -272,6 +263,44 @@ def test_find_withdirs_does_not_modify_cached_listing(self):
         assert fs.find("s3://bucket/dir") == ["bucket/dir/sub/file"]
         fs._call.assert_not_called()
 
+    def test_find_refresh_bypasses_cached_listings(self):
+        fs = self._make_fs()
+        fs.dircache[("bucket/dir", "")] = [self._file_object("dir/old")]
+        fs.dircache[("bucket/dir", "/")] = [self._file_object("dir/old")]
+        fs.dircache[("bucket/dir/sub", "/")] = [self._file_object("dir/sub/old")]
+        responses = {
+            ("dir/", ""): {"Contents": [{"Key": "dir/sub/new"}]},
+            ("dir/", "/"): {"CommonPrefixes": [{"Prefix": "dir/sub/"}]},
+            ("dir/sub/", "/"): {"Contents": [{"Key": "dir/sub/new"}]},
+        }
+        fs._call.side_effect = lambda method, **kwargs: responses[
+            (kwargs["Prefix"], kwargs["Delimiter"])
+        ]
+
+        assert fs.find("s3://bucket/dir", refresh=True) == ["bucket/dir/sub/new"]
+        # The subdirectory listings of maxdepth are refreshed as well.
+        assert fs.find("s3://bucket/dir", maxdepth=1, refresh=True) == ["bucket/dir/sub/new"]
+
+    def test_refresh_evicts_cached_object_and_bucket_not_found(self):
+        fs = self._make_fs()
+        fs.dircache["bucket/key"] = self._file_object("key")
+        fs.dircache["bucket"] = fs._directory_object("bucket", None)
+
+        def call(method, **kwargs):
+            if method in (fs._client.head_object, fs._client.head_bucket):
+                raise FileNotFoundError
+            return {}
+
+        fs._call.side_effect = call
+
+        assert fs.ls("s3://bucket/key", refresh=True) == []
+        # The next lookups must not return the deleted object and bucket from the cache.
+        assert fs.ls("s3://bucket/key") == []
+        assert not fs.exists("s3://bucket/key")
+        with pytest.raises(FileNotFoundError):
+            fs.info("s3://bucket", refresh=True)
+        assert not fs.exists("s3://bucket")
+
     def test_mkdir_creates_bucket(self):
         fs = self._make_fs()
         fs.allow_bucket_creation = True

From c98255080cb165eeb2142299e4720582e505e0e3 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 12:40:24 +0900
Subject: [PATCH 6/9] Document the prefix and refresh arguments of find()

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

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 2c3a3b279..d45cb9294 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -789,7 +789,9 @@ def find(
             maxdepth: Maximum depth to recurse (None for unlimited).
             withdirs: Whether to include directories in results (None = default behavior).
             detail: If True, return dict of {path: S3Object}; if False, return list of paths.
-            **kwargs: Additional arguments.
+            **kwargs: Additional arguments including:
+                prefix: Key prefix, relative to the path, to filter the results by.
+                refresh: If True, bypass the cache and list from S3.
 
         Returns:
             Dictionary mapping paths to S3Objects (if detail=True) or

From a95e84fe2bff7aa651ee7b7606022aad483bd963 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 12:53:48 +0900
Subject: [PATCH 7/9] Evict the bucket listing when HeadBucket finds nothing
 and qualify prefix

info() and exists() also find a bucket through the cached bucket
listing, so a deleted bucket came back from it after
info(bucket, refresh=True). Evict that listing as well, as mkdir() and
rmdir() do for buckets.

find(prefix=...) falls back to info() of the path when nothing is
listed, which ignores the prefix. Say so in the docstring.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py            | 6 +++++-
 tests/pyathena/filesystem/test_s3.py | 1 +
 2 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index d45cb9294..b68fb3824 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -316,6 +316,8 @@ def _head_bucket(self, bucket, refresh: bool = False) -> S3Object | None:
                 )
             except FileNotFoundError:
                 self.dircache.pop(bucket, None)
+                # The cached bucket listing may still contain the bucket.
+                self.dircache.pop("", None)
                 return None
             file = S3Object(
                 init={
@@ -790,7 +792,9 @@ def find(
             withdirs: Whether to include directories in results (None = default behavior).
             detail: If True, return dict of {path: S3Object}; if False, return list of paths.
             **kwargs: Additional arguments including:
-                prefix: Key prefix, relative to the path, to filter the results by.
+                prefix: Key prefix, relative to the path, to filter the listed keys
+                    by. If nothing is listed and the path itself is an object, that
+                    object is returned regardless of the prefix.
                 refresh: If True, bypass the cache and list from S3.
 
         Returns:
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 21ed555c6..3a520f78b 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -285,6 +285,7 @@ def test_refresh_evicts_cached_object_and_bucket_not_found(self):
         fs = self._make_fs()
         fs.dircache["bucket/key"] = self._file_object("key")
         fs.dircache["bucket"] = fs._directory_object("bucket", None)
+        fs.dircache[""] = [fs._directory_object("bucket", None)]
 
         def call(method, **kwargs):
             if method in (fs._client.head_object, fs._client.head_bucket):

From 51a43622104c09789063c5a0cc026ed3bb1c5db6 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 12:56:58 +0900
Subject: [PATCH 8/9] Honor refresh in exists()

exists() ignored its keyword arguments, so exists(path, refresh=True)
answered from the cache. fsspec's exists() passes them to info(), which
honors refresh. Skip the cache lookups and pass refresh to info() and
HeadBucket instead.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py            | 18 ++++++++++--------
 tests/pyathena/filesystem/test_s3.py | 20 ++++++++++++++++++++
 2 files changed, 30 insertions(+), 8 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index b68fb3824..3f9021044 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -819,7 +819,8 @@ def exists(self, path: str, **kwargs) -> bool:
 
         Args:
             path: S3 path to check (e.g., "s3://bucket" or "s3://bucket/key").
-            **kwargs: Additional arguments (unused).
+            **kwargs: Additional arguments including:
+                refresh: If True, bypass the cache and query S3.
 
         Returns:
             True if the path exists, False otherwise.
@@ -829,6 +830,7 @@ def exists(self, path: str, **kwargs) -> bool:
             >>> fs.exists("s3://my-bucket/file.txt")
             >>> fs.exists("s3://my-bucket/")
         """
+        refresh = kwargs.pop("refresh", False)
         path = self._strip_protocol(path)
         if path in ["", "/"]:
             # The root always exists.
@@ -836,22 +838,22 @@ def exists(self, path: str, **kwargs) -> bool:
         bucket, key, _ = self.parse_path(path)
         if key:
             try:
-                if self._ls_from_cache(path):
+                if not refresh and self._ls_from_cache(path):
                     return True
-                info = self.info(path)
+                info = self.info(path, refresh=refresh)
                 return bool(info)
             except FileNotFoundError:
                 return False
-        elif self.dircache.get(bucket, False):
-            return True
-        else:
+        if not refresh:
+            if self.dircache.get(bucket, False):
+                return True
             try:
                 if self._ls_from_cache(bucket):
                     return True
             except FileNotFoundError:
                 pass
-            file = self._head_bucket(bucket)
-            return bool(file)
+        file = self._head_bucket(bucket, refresh=refresh)
+        return bool(file)
 
     def rm_file(self, path: str, **kwargs) -> None:
         """Delete an S3 object with DeleteObject.
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 3a520f78b..26f11caad 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -302,6 +302,26 @@ def call(method, **kwargs):
             fs.info("s3://bucket", refresh=True)
         assert not fs.exists("s3://bucket")
 
+    def test_exists_refresh_bypasses_cache(self):
+        fs = self._make_fs()
+        fs.dircache["bucket/key"] = self._file_object("key")
+        fs.dircache["bucket"] = fs._directory_object("bucket", None)
+        fs.dircache[""] = [fs._directory_object("bucket", None)]
+
+        def call(method, **kwargs):
+            if method in (fs._client.head_object, fs._client.head_bucket):
+                raise FileNotFoundError
+            return {}
+
+        fs._call.side_effect = call
+
+        assert fs.exists("s3://bucket/key")
+        assert fs.exists("s3://bucket")
+        fs._call.assert_not_called()
+
+        assert not fs.exists("s3://bucket/key", refresh=True)
+        assert not fs.exists("s3://bucket", refresh=True)
+
     def test_mkdir_creates_bucket(self):
         fs = self._make_fs()
         fs.allow_bucket_creation = True

From b16c58f484bedd1f6976f3d1f0100286ca76c064 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 13:06:22 +0900
Subject: [PATCH 9/9] Keep the bucket listing when a bucket not in it is
 missing

Evicting the cached bucket listing on every missing bucket made the
next exists() of other listed buckets issue HeadBucket, which raises
PermissionError for callers allowed only to list buckets. Evict it only
when it still lists the missing bucket. Also note that the find()
prefix fallback to the path itself applies only without maxdepth.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py            | 10 ++++++----
 tests/pyathena/filesystem/test_s3.py | 11 +++++++++++
 2 files changed, 17 insertions(+), 4 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 3f9021044..940a2649e 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -316,8 +316,10 @@ def _head_bucket(self, bucket, refresh: bool = False) -> S3Object | None:
                 )
             except FileNotFoundError:
                 self.dircache.pop(bucket, None)
-                # The cached bucket listing may still contain the bucket.
-                self.dircache.pop("", None)
+                # Evict the cached bucket listing only if it still lists the bucket.
+                buckets = self.dircache.get("")
+                if buckets and any(b.name == bucket for b in buckets):
+                    self.dircache.pop("", None)
                 return None
             file = S3Object(
                 init={
@@ -793,8 +795,8 @@ def find(
             detail: If True, return dict of {path: S3Object}; if False, return list of paths.
             **kwargs: Additional arguments including:
                 prefix: Key prefix, relative to the path, to filter the listed keys
-                    by. If nothing is listed and the path itself is an object, that
-                    object is returned regardless of the prefix.
+                    by. Without maxdepth, if nothing is listed and the path itself is
+                    an object, that object is returned regardless of the prefix.
                 refresh: If True, bypass the cache and list from S3.
 
         Returns:
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 26f11caad..ed6335019 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -322,6 +322,17 @@ def call(method, **kwargs):
         assert not fs.exists("s3://bucket/key", refresh=True)
         assert not fs.exists("s3://bucket", refresh=True)
 
+    def test_missing_bucket_keeps_bucket_listing_without_it(self):
+        fs = self._make_fs()
+        fs.dircache[""] = [fs._directory_object("bucket", None)]
+        fs._call.side_effect = FileNotFoundError
+
+        assert not fs.exists("s3://missing")
+        # Other buckets are still answered from the cached bucket listing.
+        fs._call.reset_mock()
+        assert fs.exists("s3://bucket")
+        fs._call.assert_not_called()
+
     def test_mkdir_creates_bucket(self):
         fs = self._make_fs()
         fs.allow_bucket_creation = True