From 688b087d6a95c35237e11743f167601a246cfda9 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 17:09:43 +0900
Subject: [PATCH 01/12] Align S3FileSystem.find() with fsspec for the path
 itself and prefix levels

With withdirs, find() now includes the path itself when it is a
directory, as fsspec does for posix glob compliance, so glob("d/**")
matches d (#963). The root is known from a non-empty listing, so this
costs no extra request.

With maxdepth, each slash in prefix now counts as one level from the
path, and with withdirs both branches include only the directories
whose relative paths start with the prefix (#964).

With maxdepth, an object path now returns the object itself when
nothing is listed, regardless of prefix, as without maxdepth and as in
fsspec (#966).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py                  | 124 ++++++++++++-------
 tests/pyathena/filesystem/test_s3.py       | 137 ++++++++++++++++++++-
 tests/pyathena/filesystem/test_s3_async.py |  17 +++
 3 files changed, 230 insertions(+), 48 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index eceb61c7..b67a7dd5 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -829,7 +829,7 @@ def info(self, path: str, **kwargs) -> S3Object:
         raise FileNotFoundError(path)
 
     def _extract_parent_directories(
-        self, files: list[S3Object], bucket: str, base_key: str | None
+        self, files: list[S3Object], bucket: str, base_key: str | None, prefix: str = ""
     ) -> list[S3Object]:
         """Extract parent directory objects from file paths.
 
@@ -840,6 +840,8 @@ def _extract_parent_directories(
             files: List of S3Object instances representing files.
             bucket: S3 bucket name.
             base_key: Base key path to calculate relative paths from.
+            prefix: Key prefix, relative to the base key, that the paths of the
+                directories relative to the base key must start with.
 
         Returns:
             List of S3Object instances representing directories.
@@ -861,11 +863,9 @@ def _extract_parent_directories(
                 # Get all parent directories
                 parts = relative_path.split("/")
                 for i in range(1, len(parts)):
-                    if base_key:
-                        dir_path = base_key + "/" + "/".join(parts[:i])
-                    else:
-                        dir_path = "/".join(parts[:i])
-                    dirs.add(dir_path)
+                    relative_dir = "/".join(parts[:i])
+                    if relative_dir.startswith(prefix):
+                        dirs.add(f"{base_key}/{relative_dir}" if base_key else relative_dir)
 
         return [self._directory_object(bucket, dir_path) for dir_path in dirs]
 
@@ -899,54 +899,81 @@ 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)
+        refresh = kwargs.pop("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="/", refresh=refresh)
-
-            for item in current_items:
-                if item.type == S3ObjectType.S3_OBJECT_TYPE_FILE:
-                    # Add files
-                    result.append(item)
-                elif item.type == S3ObjectType.S3_OBJECT_TYPE_DIRECTORY:
-                    # Add directory if withdirs is True
-                    if withdirs:
-                        result.append(item)
-
-                    # Recursively explore subdirectory if depth allows
-                    if maxdepth > 1:
-                        sub_path = f"s3://{bucket}/{item.key}"
-                        sub_results = self._find(
-                            sub_path, maxdepth=maxdepth - 1, withdirs=withdirs, **kwargs
-                        )
-                        result.extend(sub_results)
-
-            return result
-
-        # For unlimited depth, use the original approach (get all files at once)
-        files = self._ls_dirs(path, prefix=prefix, delimiter="", refresh=refresh)
-        if not files and key:
+            # The entries listed with the prefix lie as many levels further
+            # below the path as the prefix has slashes.
+            levels = maxdepth - prefix.count("/")
+            files = (
+                self._find_levels(path, levels, withdirs, prefix=prefix, refresh=refresh)
+                if levels >= 1
+                else []
+            )
+        else:
+            files = self._ls_dirs(path, prefix=prefix, delimiter="", refresh=refresh)
+            # S3 doesn't return directory entries without a delimiter, so the
+            # directories are derived from the listed keys.
+            if withdirs:
+                # Build a new list; files may be the cached listing.
+                files = files + self._extract_parent_directories(files, bucket, key, prefix)
+
+        if files:
+            # Something is listed below the path, so the path is a directory,
+            # which fsspec includes with the directories.
+            if withdirs:
+                files = [self._directory_object(bucket, key), *files]
+        elif key:
+            # As in fsspec, the path itself is returned if it is an object.
             try:
                 files = [self.info(path, refresh=refresh)]
             except FileNotFoundError:
                 files = []
 
-        # If withdirs is True, we need to derive directories from file paths
-        if withdirs:
-            # 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:
+        if not withdirs:
             files = [f for f in files if f.type != S3ObjectType.S3_OBJECT_TYPE_DIRECTORY]
-
         return files
 
+    def _find_levels(
+        self,
+        path: str,
+        maxdepth: int,
+        withdirs: bool | None,
+        prefix: str = "",
+        refresh: bool = False,
+    ) -> list[S3Object]:
+        """List the objects below a path level by level with ``Delimiter="/"``.
+
+        Args:
+            path: S3 path to search under.
+            maxdepth: Number of levels to list, at least 1.
+            withdirs: Whether to include directories in the result.
+            prefix: Key prefix, relative to the path, to filter the first
+                level by.
+            refresh: If True, bypass the cache and list from S3.
+
+        Returns:
+            The objects found, and the directories if ``withdirs`` is True.
+        """
+        bucket, _, _ = self.parse_path(path)
+        result: list[S3Object] = []
+        for item in self._ls_dirs(path, prefix=prefix, delimiter="/", refresh=refresh):
+            if item.type == S3ObjectType.S3_OBJECT_TYPE_FILE:
+                result.append(item)
+            elif item.type == S3ObjectType.S3_OBJECT_TYPE_DIRECTORY:
+                if withdirs:
+                    result.append(item)
+                if maxdepth > 1:
+                    result.extend(
+                        self._find_levels(
+                            f"s3://{bucket}/{item.key}",
+                            maxdepth - 1,
+                            withdirs,
+                            refresh=refresh,
+                        )
+                    )
+        return result
+
     def find(
         self,
         path: str,
@@ -959,7 +986,9 @@ def find(
 
         Recursively searches for files under the specified path, with optional
         depth limiting and directory inclusion. Uses efficient S3 list operations
-        with delimiter handling for performance.
+        with delimiter handling for performance. As in fsspec, the result
+        includes the path itself if it is a directory and withdirs is True, or
+        if it is an object and nothing is listed below it.
 
         Args:
             path: S3 path to search under (e.g., "s3://bucket/prefix").
@@ -970,8 +999,9 @@ 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. Without maxdepth, if nothing is listed and the path itself is
-                    an object, that object is returned regardless of the prefix.
+                    by. Each slash in the prefix counts as one level of maxdepth.
+                    With withdirs, only the directories whose paths relative to the
+                    path start with the prefix are included.
                 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 a0948219..21486162 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -26,6 +26,7 @@
 from fsspec import Callback
 from fsspec.dircache import DirCache
 from fsspec.implementations.dirfs import DirFileSystem
+from fsspec.implementations.memory import MemoryFileSystem
 
 import pyathena
 from pyathena.filesystem import register_s3_filesystem
@@ -735,7 +736,7 @@ def test_find_withdirs_does_not_modify_cached_listing(self):
         fs = self._make_fs()
         fs.dircache[("bucket/dir", "")] = [self._file_object("dir/sub/file")]
 
-        expected = ["bucket/dir/sub", "bucket/dir/sub/file"]
+        expected = ["bucket/dir", "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"]
@@ -780,6 +781,7 @@ def test_find_maxdepth_counts_levels_like_fsspec(self):
 
         assert fs.find("s3://bucket/dir", maxdepth=1) == ["bucket/dir/direct"]
         assert sorted(fs.find("s3://bucket/dir", maxdepth=1, withdirs=True)) == [
+            "bucket/dir",
             "bucket/dir/direct",
             "bucket/dir/sub",
         ]
@@ -793,6 +795,124 @@ def test_find_maxdepth_counts_levels_like_fsspec(self):
             "bucket/dir/sub/nested",
         ]
 
+    FIND_KEYS = ("dir/direct", "dir/sub/nested", "dir/sub/deep/file")
+
+    @staticmethod
+    def _serve_keys(fs, keys):
+        # Answer the ListObjectsV2 and HeadObject requests of "bucket" from
+        # the given keys.
+        def call(method, **kwargs):
+            if method is fs._client.head_object:
+                if kwargs["Key"] not in keys:
+                    raise FileNotFoundError(kwargs["Key"])
+                return {"ContentLength": 0}
+            prefix, delimiter = kwargs["Prefix"], kwargs["Delimiter"]
+            contents, prefixes = [], set()
+            for key in sorted(keys):
+                if not key.startswith(prefix):
+                    continue
+                rest = key[len(prefix) :]
+                if delimiter and delimiter in rest:
+                    prefixes.add(prefix + rest.split(delimiter)[0] + delimiter)
+                else:
+                    contents.append({"Key": key})
+            return {
+                "Contents": contents,
+                "CommonPrefixes": [{"Prefix": p} for p in sorted(prefixes)],
+                "KeyCount": len(contents) + len(prefixes),
+            }
+
+        fs._call.side_effect = call
+
+    @staticmethod
+    def _memory_fs(keys):
+        # Build an fsspec MemoryFileSystem with the keys under "/bucket".
+        memory = MemoryFileSystem(skip_instance_cache=True)
+        memory.store = {}
+        memory.pseudo_dirs = [""]
+        for key in keys:
+            memory.pipe(f"/bucket/{key}", b"")
+        return memory
+
+    @pytest.mark.parametrize(
+        ("path", "maxdepth", "withdirs"),
+        [
+            ("dir", None, True),
+            ("dir", None, False),
+            ("dir", 1, True),
+            ("dir", 2, True),
+            ("dir", 1, False),
+            ("dir/sub", 1, True),
+            ("dir/direct", None, True),
+            ("dir/direct", 1, True),
+            ("dir/direct", 1, False),
+            ("missing", None, True),
+            ("missing", 1, True),
+        ],
+    )
+    def test_find_matches_fsspec(self, path, maxdepth, withdirs):
+        fs = self._make_fs()
+        self._serve_keys(fs, self.FIND_KEYS)
+        memory = self._memory_fs(self.FIND_KEYS)
+
+        expected = [p.lstrip("/") for p in memory.find(f"/bucket/{path}", maxdepth, withdirs)]
+        assert sorted(fs.find(f"s3://bucket/{path}", maxdepth, withdirs)) == expected
+
+    @pytest.mark.parametrize("pattern", ["dir/**", "dir/*", "dir/*/*", "dir/s*", "dir/**/file"])
+    def test_glob_matches_fsspec(self, pattern):
+        fs = self._make_fs()
+        self._serve_keys(fs, self.FIND_KEYS)
+        memory = self._memory_fs(self.FIND_KEYS)
+
+        expected = [p.lstrip("/") for p in memory.glob(f"/bucket/{pattern}")]
+        assert sorted(fs.glob(f"s3://bucket/{pattern}")) == expected
+
+    def test_find_withdirs_lists_root_without_extra_requests(self):
+        fs = self._make_fs()
+        self._serve_keys(fs, self.FIND_KEYS)
+
+        assert "bucket/dir" in fs.find("s3://bucket/dir", withdirs=True)
+        assert fs._call.call_count == 1
+        fs._call.reset_mock()
+        assert "bucket/dir" in fs.find("s3://bucket/dir", maxdepth=1, withdirs=True)
+        assert fs._call.call_count == 1
+
+    def test_find_prefix_counts_levels_from_path(self):
+        fs = self._make_fs()
+        self._serve_keys(fs, self.FIND_KEYS)
+
+        # Nothing directly under dir/ starts with "sub/deep/".
+        assert fs.find("s3://bucket/dir", maxdepth=1, prefix="sub/deep/") == []
+        assert fs.find("s3://bucket/dir", maxdepth=2, prefix="sub/deep/") == []
+        assert fs.find("s3://bucket/dir", maxdepth=3, prefix="sub/deep/") == [
+            "bucket/dir/sub/deep/file"
+        ]
+        assert sorted(fs.find("s3://bucket/dir", maxdepth=2, prefix="sub/")) == [
+            "bucket/dir/sub/nested"
+        ]
+        # The directories above the prefix do not start with it, with or
+        # without maxdepth.
+        expected = [
+            "bucket/dir",
+            "bucket/dir/sub/deep",
+            "bucket/dir/sub/deep/file",
+            "bucket/dir/sub/nested",
+        ]
+        assert sorted(fs.find("s3://bucket/dir", maxdepth=3, prefix="sub/", withdirs=True)) == (
+            expected
+        )
+        assert sorted(fs.find("s3://bucket/dir", prefix="sub/", withdirs=True)) == expected
+
+    @pytest.mark.parametrize("maxdepth", [None, 1])
+    def test_find_object_path_ignores_prefix(self, maxdepth):
+        fs = self._make_fs()
+        self._serve_keys(fs, self.FIND_KEYS)
+
+        # As in fsspec, the object itself is returned when nothing is listed.
+        assert fs.find("s3://bucket/dir/direct", maxdepth=maxdepth, prefix="x") == [
+            "bucket/dir/direct"
+        ]
+
     def test_refresh_evicts_cached_object_and_bucket_not_found(self):
         fs = self._make_fs()
         fs.dircache["bucket/key"] = self._file_object("key")
@@ -3433,6 +3553,15 @@ def test_find_maxdepth(self, fs):
         result = fs.find(dir_)
         assert len(result) == 4
 
+        # Each slash in the prefix counts as one level
+        assert fs.find(dir_, maxdepth=1, prefix="level1/") == []
+        assert fs.find(dir_, maxdepth=2, prefix="level1/") == [
+            fs._strip_protocol(f"{dir_}/level1/file1.txt")
+        ]
+
+        # An object path returns the object itself
+        assert fs.find(f"{dir_}/file0.txt", maxdepth=1) == [fs._strip_protocol(f"{dir_}/file0.txt")]
+
     def test_find_withdirs(self, fs):
         dir_ = f"s3://{ENV.s3_staging_bucket}/{ENV.s3_staging_key}{ENV.schema}/filesystem/test_find_withdirs"
         # Create directory structure with files
@@ -3450,6 +3579,8 @@ def test_find_withdirs(self, fs):
         # Test withdirs=True
         result = fs.find(dir_, withdirs=True)
         assert len(result) > 4  # Files and directories
+        assert fs._strip_protocol(dir_) in result
+        assert fs._strip_protocol(dir_) in fs.find(dir_, maxdepth=1, withdirs=True)
 
         # Verify directories are included
         dirs = [r for r in result if not r.endswith(".txt")]
@@ -3478,6 +3609,10 @@ def test_glob(self, fs):
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/nested/*")
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/nested/test_*")
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/*/*")
+        assert fs.glob(f"{dir_}/nested/**") == [
+            fs._strip_protocol(f"{dir_}/nested"),
+            fs._strip_protocol(path),
+        ]
 
         with pytest.raises(ValueError):  # noqa: PT011
             fs.glob("*")
diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py
index a5b2367b..524f2734 100644
--- a/tests/pyathena/filesystem/test_s3_async.py
+++ b/tests/pyathena/filesystem/test_s3_async.py
@@ -1004,6 +1004,17 @@ async def test_find_maxdepth(self, fs):
         result = await fs._find(dir_)
         assert len(result) == 4
 
+        # Each slash in the prefix counts as one level
+        assert await fs._find(dir_, maxdepth=1, prefix="level1/") == []
+        assert await fs._find(dir_, maxdepth=2, prefix="level1/") == [
+            fs._strip_protocol(f"{dir_}/level1/file1.txt")
+        ]
+
+        # An object path returns the object itself
+        assert await fs._find(f"{dir_}/file0.txt", maxdepth=1) == [
+            fs._strip_protocol(f"{dir_}/file0.txt")
+        ]
+
     @pytest.mark.asyncio
     async def test_find_withdirs(self, fs):
         dir_ = (
@@ -1025,6 +1036,8 @@ async def test_find_withdirs(self, fs):
         # Test withdirs=True
         result = await fs._find(dir_, withdirs=True)
         assert len(result) > 4  # Files and directories
+        assert fs._strip_protocol(dir_) in result
+        assert fs._strip_protocol(dir_) in await fs._find(dir_, maxdepth=1, withdirs=True)
 
         # Verify directories are included
         dirs = [r for r in result if not r.endswith(".txt")]
@@ -1057,6 +1070,10 @@ async def test_glob(self, fs):
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/nested/*")
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/nested/test_*")
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/*/*")
+        assert fs.glob(f"{dir_}/nested/**") == [
+            fs._strip_protocol(f"{dir_}/nested"),
+            fs._strip_protocol(path),
+        ]
 
         with pytest.raises(ValueError):  # noqa: PT011
             fs.glob("*")

From 579c12061b5dbcb3aefd03f5a8506411d8a1f96c Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 17:15:50 +0900
Subject: [PATCH 02/12] Describe the empty-bucket case of find() and compare
 glob maxdepth with fsspec

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

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index b67a7dd5..b1bb9a28 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -987,8 +987,8 @@ def find(
         Recursively searches for files under the specified path, with optional
         depth limiting and directory inclusion. Uses efficient S3 list operations
         with delimiter handling for performance. As in fsspec, the result
-        includes the path itself if it is a directory and withdirs is True, or
-        if it is an object and nothing is listed below it.
+        includes the path itself if withdirs is True and objects exist below
+        it, or if it is an object and nothing is listed below it.
 
         Args:
             path: S3 path to search under (e.g., "s3://bucket/prefix").
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 21486162..638f4b55 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -858,14 +858,26 @@ def test_find_matches_fsspec(self, path, maxdepth, withdirs):
         expected = [p.lstrip("/") for p in memory.find(f"/bucket/{path}", maxdepth, withdirs)]
         assert sorted(fs.find(f"s3://bucket/{path}", maxdepth, withdirs)) == expected
 
-    @pytest.mark.parametrize("pattern", ["dir/**", "dir/*", "dir/*/*", "dir/s*", "dir/**/file"])
-    def test_glob_matches_fsspec(self, pattern):
+    @pytest.mark.parametrize(
+        ("pattern", "maxdepth"),
+        [
+            ("dir/**", None),
+            ("dir/**", 1),
+            ("dir/**", 2),
+            ("dir/*", None),
+            ("dir/*/*", None),
+            ("dir/s*", None),
+            ("dir/**/file", None),
+            ("missing/*", None),
+        ],
+    )
+    def test_glob_matches_fsspec(self, pattern, maxdepth):
         fs = self._make_fs()
         self._serve_keys(fs, self.FIND_KEYS)
         memory = self._memory_fs(self.FIND_KEYS)
 
-        expected = [p.lstrip("/") for p in memory.glob(f"/bucket/{pattern}")]
-        assert sorted(fs.glob(f"s3://bucket/{pattern}")) == expected
+        expected = [p.lstrip("/") for p in memory.glob(f"/bucket/{pattern}", maxdepth=maxdepth)]
+        assert sorted(fs.glob(f"s3://bucket/{pattern}", maxdepth=maxdepth)) == expected
 
     def test_find_withdirs_lists_root_without_extra_requests(self):
         fs = self._make_fs()

From 535bd1702bd7c03aefca48af755a5ef894975b66 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 17:21:23 +0900
Subject: [PATCH 03/12] Keep subdirectories in the find() listing check without
 withdirs

_find_levels() dropped directories when withdirs was false, so a path
with only subdirectories below it looked empty and was looked up with
info() (HeadObject and ListObjectsV2). It now returns all entries and
_find() drops the directories at the end.

The live glob checks no longer assume that the test directory holds
only this run's file, which a rerun in the same session breaks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py                  | 33 ++++++----------------
 tests/pyathena/filesystem/test_s3.py       | 12 ++++----
 tests/pyathena/filesystem/test_s3_async.py |  5 +---
 3 files changed, 17 insertions(+), 33 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index b1bb9a28..b519c518 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -906,7 +906,7 @@ def _find(
             # below the path as the prefix has slashes.
             levels = maxdepth - prefix.count("/")
             files = (
-                self._find_levels(path, levels, withdirs, prefix=prefix, refresh=refresh)
+                self._find_levels(path, levels, prefix=prefix, refresh=refresh)
                 if levels >= 1
                 else []
             )
@@ -935,43 +935,28 @@ def _find(
         return files
 
     def _find_levels(
-        self,
-        path: str,
-        maxdepth: int,
-        withdirs: bool | None,
-        prefix: str = "",
-        refresh: bool = False,
+        self, path: str, maxdepth: int, prefix: str = "", refresh: bool = False
     ) -> list[S3Object]:
-        """List the objects below a path level by level with ``Delimiter="/"``.
+        """List the entries below a path level by level with ``Delimiter="/"``.
 
         Args:
             path: S3 path to search under.
             maxdepth: Number of levels to list, at least 1.
-            withdirs: Whether to include directories in the result.
             prefix: Key prefix, relative to the path, to filter the first
                 level by.
             refresh: If True, bypass the cache and list from S3.
 
         Returns:
-            The objects found, and the directories if ``withdirs`` is True.
+            The objects and directories found.
         """
         bucket, _, _ = self.parse_path(path)
         result: list[S3Object] = []
         for item in self._ls_dirs(path, prefix=prefix, delimiter="/", refresh=refresh):
-            if item.type == S3ObjectType.S3_OBJECT_TYPE_FILE:
-                result.append(item)
-            elif item.type == S3ObjectType.S3_OBJECT_TYPE_DIRECTORY:
-                if withdirs:
-                    result.append(item)
-                if maxdepth > 1:
-                    result.extend(
-                        self._find_levels(
-                            f"s3://{bucket}/{item.key}",
-                            maxdepth - 1,
-                            withdirs,
-                            refresh=refresh,
-                        )
-                    )
+            result.append(item)
+            if item.type == S3ObjectType.S3_OBJECT_TYPE_DIRECTORY and maxdepth > 1:
+                result.extend(
+                    self._find_levels(f"s3://{bucket}/{item.key}", maxdepth - 1, refresh=refresh)
+                )
         return result
 
     def find(
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 638f4b55..e1053144 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -879,7 +879,7 @@ def test_glob_matches_fsspec(self, pattern, maxdepth):
         expected = [p.lstrip("/") for p in memory.glob(f"/bucket/{pattern}", maxdepth=maxdepth)]
         assert sorted(fs.glob(f"s3://bucket/{pattern}", maxdepth=maxdepth)) == expected
 
-    def test_find_withdirs_lists_root_without_extra_requests(self):
+    def test_find_directory_without_extra_requests(self):
         fs = self._make_fs()
         self._serve_keys(fs, self.FIND_KEYS)
 
@@ -888,6 +888,11 @@ def test_find_withdirs_lists_root_without_extra_requests(self):
         fs._call.reset_mock()
         assert "bucket/dir" in fs.find("s3://bucket/dir", maxdepth=1, withdirs=True)
         assert fs._call.call_count == 1
+        # Only a subdirectory is listed; it is dropped without withdirs, but
+        # the path is a directory, so it is not looked up as an object.
+        fs._call.reset_mock()
+        assert fs.find("s3://bucket/dir", maxdepth=1, prefix="s") == []
+        assert fs._call.call_count == 1
 
     def test_find_prefix_counts_levels_from_path(self):
         fs = self._make_fs()
@@ -3621,10 +3626,7 @@ def test_glob(self, fs):
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/nested/*")
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/nested/test_*")
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/*/*")
-        assert fs.glob(f"{dir_}/nested/**") == [
-            fs._strip_protocol(f"{dir_}/nested"),
-            fs._strip_protocol(path),
-        ]
+        assert fs._strip_protocol(f"{dir_}/nested") in fs.glob(f"{dir_}/nested/**")
 
         with pytest.raises(ValueError):  # noqa: PT011
             fs.glob("*")
diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py
index 524f2734..92e36fa6 100644
--- a/tests/pyathena/filesystem/test_s3_async.py
+++ b/tests/pyathena/filesystem/test_s3_async.py
@@ -1070,10 +1070,7 @@ async def test_glob(self, fs):
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/nested/*")
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/nested/test_*")
         assert fs._strip_protocol(path) in fs.glob(f"{dir_}/*/*")
-        assert fs.glob(f"{dir_}/nested/**") == [
-            fs._strip_protocol(f"{dir_}/nested"),
-            fs._strip_protocol(path),
-        ]
+        assert fs._strip_protocol(f"{dir_}/nested") in fs.glob(f"{dir_}/nested/**")
 
         with pytest.raises(ValueError):  # noqa: PT011
             fs.glob("*")

From 3c882ce5b44478fd81616d774c9396bd5f23a138 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 17:26:15 +0900
Subject: [PATCH 04/12] List find() subdirectories under stripped paths and
 simplify the levels

_find_levels() recursed with s3:// paths, so the subdirectory listings
were cached under keys that invalidate_cache() never drops and ls()
never reads. It now recurses with the entry names, as the recursion
through _find() did before.

The unlimited branch derives the directories below the last slash of
the prefix by passing that as the base key, so
_extract_parent_directories() keeps its signature. The maxdepth test
uses the shared fake listing, and the async live tests keep only the
glob check, since AioS3FileSystem._find() runs the sync _find().

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py                  | 36 ++++++-----
 tests/pyathena/filesystem/test_s3.py       | 71 +++++++++++-----------
 tests/pyathena/filesystem/test_s3_async.py | 13 ----
 3 files changed, 53 insertions(+), 67 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index b519c518..246b25f1 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -829,7 +829,7 @@ def info(self, path: str, **kwargs) -> S3Object:
         raise FileNotFoundError(path)
 
     def _extract_parent_directories(
-        self, files: list[S3Object], bucket: str, base_key: str | None, prefix: str = ""
+        self, files: list[S3Object], bucket: str, base_key: str | None
     ) -> list[S3Object]:
         """Extract parent directory objects from file paths.
 
@@ -840,8 +840,6 @@ def _extract_parent_directories(
             files: List of S3Object instances representing files.
             bucket: S3 bucket name.
             base_key: Base key path to calculate relative paths from.
-            prefix: Key prefix, relative to the base key, that the paths of the
-                directories relative to the base key must start with.
 
         Returns:
             List of S3Object instances representing directories.
@@ -863,9 +861,11 @@ def _extract_parent_directories(
                 # Get all parent directories
                 parts = relative_path.split("/")
                 for i in range(1, len(parts)):
-                    relative_dir = "/".join(parts[:i])
-                    if relative_dir.startswith(prefix):
-                        dirs.add(f"{base_key}/{relative_dir}" if base_key else relative_dir)
+                    if base_key:
+                        dir_path = base_key + "/" + "/".join(parts[:i])
+                    else:
+                        dir_path = "/".join(parts[:i])
+                    dirs.add(dir_path)
 
         return [self._directory_object(bucket, dir_path) for dir_path in dirs]
 
@@ -904,19 +904,18 @@ def _find(
         if maxdepth is not None:
             # The entries listed with the prefix lie as many levels further
             # below the path as the prefix has slashes.
-            levels = maxdepth - prefix.count("/")
-            files = (
-                self._find_levels(path, levels, prefix=prefix, refresh=refresh)
-                if levels >= 1
-                else []
+            files = self._find_levels(
+                path, maxdepth - prefix.count("/"), prefix=prefix, refresh=refresh
             )
         else:
             files = self._ls_dirs(path, prefix=prefix, delimiter="", refresh=refresh)
             # S3 doesn't return directory entries without a delimiter, so the
-            # directories are derived from the listed keys.
+            # directories are derived from the listed keys, below the last
+            # slash of the prefix, as with maxdepth.
             if withdirs:
+                base_key = "/".join(k for k in (key, prefix.rpartition("/")[0]) if k)
                 # Build a new list; files may be the cached listing.
-                files = files + self._extract_parent_directories(files, bucket, key, prefix)
+                files = files + self._extract_parent_directories(files, bucket, base_key)
 
         if files:
             # Something is listed below the path, so the path is a directory,
@@ -941,7 +940,7 @@ def _find_levels(
 
         Args:
             path: S3 path to search under.
-            maxdepth: Number of levels to list, at least 1.
+            maxdepth: Number of levels to list.
             prefix: Key prefix, relative to the path, to filter the first
                 level by.
             refresh: If True, bypass the cache and list from S3.
@@ -949,14 +948,13 @@ def _find_levels(
         Returns:
             The objects and directories found.
         """
-        bucket, _, _ = self.parse_path(path)
+        if maxdepth < 1:
+            return []
         result: list[S3Object] = []
         for item in self._ls_dirs(path, prefix=prefix, delimiter="/", refresh=refresh):
             result.append(item)
-            if item.type == S3ObjectType.S3_OBJECT_TYPE_DIRECTORY and maxdepth > 1:
-                result.extend(
-                    self._find_levels(f"s3://{bucket}/{item.key}", maxdepth - 1, refresh=refresh)
-                )
+            if item.type == S3ObjectType.S3_OBJECT_TYPE_DIRECTORY:
+                result.extend(self._find_levels(item.name, maxdepth - 1, refresh=refresh))
         return result
 
     def find(
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index e1053144..659e5f2f 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -760,41 +760,6 @@ def test_find_refresh_bypasses_cached_listings(self):
         # The subdirectory listings of maxdepth are refreshed as well.
         assert fs.find("s3://bucket/dir", maxdepth=2, refresh=True) == ["bucket/dir/sub/new"]
 
-    def test_find_maxdepth_counts_levels_like_fsspec(self):
-        fs = self._make_fs()
-        responses = {
-            "dir/": {
-                "Contents": [{"Key": "dir/direct"}],
-                "CommonPrefixes": [{"Prefix": "dir/sub/"}],
-            },
-            "dir/sub/": {
-                "Contents": [{"Key": "dir/sub/nested"}],
-                "CommonPrefixes": [{"Prefix": "dir/sub/deep/"}],
-            },
-            "dir/sub/deep/": {"Contents": [{"Key": "dir/sub/deep/file"}]},
-        }
-        fs._call.side_effect = lambda method, **kwargs: responses[kwargs["Prefix"]]
-
-        with pytest.raises(ValueError, match="maxdepth must be at least 1"):
-            fs.find("s3://bucket/dir", maxdepth=0)
-        fs._call.assert_not_called()
-
-        assert fs.find("s3://bucket/dir", maxdepth=1) == ["bucket/dir/direct"]
-        assert sorted(fs.find("s3://bucket/dir", maxdepth=1, withdirs=True)) == [
-            "bucket/dir",
-            "bucket/dir/direct",
-            "bucket/dir/sub",
-        ]
-        assert sorted(fs.find("s3://bucket/dir", maxdepth=2)) == [
-            "bucket/dir/direct",
-            "bucket/dir/sub/nested",
-        ]
-        assert sorted(fs.find("s3://bucket/dir", maxdepth=3)) == [
-            "bucket/dir/direct",
-            "bucket/dir/sub/deep/file",
-            "bucket/dir/sub/nested",
-        ]
-
     FIND_KEYS = ("dir/direct", "dir/sub/nested", "dir/sub/deep/file")
 
     @staticmethod
@@ -834,6 +799,25 @@ def _memory_fs(keys):
             memory.pipe(f"/bucket/{key}", b"")
         return memory
 
+    def test_find_maxdepth_counts_levels_like_fsspec(self):
+        fs = self._make_fs()
+        self._serve_keys(fs, self.FIND_KEYS)
+
+        with pytest.raises(ValueError, match="maxdepth must be at least 1"):
+            fs.find("s3://bucket/dir", maxdepth=0)
+        fs._call.assert_not_called()
+
+        assert fs.find("s3://bucket/dir", maxdepth=1) == ["bucket/dir/direct"]
+        assert sorted(fs.find("s3://bucket/dir", maxdepth=2)) == [
+            "bucket/dir/direct",
+            "bucket/dir/sub/nested",
+        ]
+        assert sorted(fs.find("s3://bucket/dir", maxdepth=3)) == [
+            "bucket/dir/direct",
+            "bucket/dir/sub/deep/file",
+            "bucket/dir/sub/nested",
+        ]
+
     @pytest.mark.parametrize(
         ("path", "maxdepth", "withdirs"),
         [
@@ -894,6 +878,23 @@ def test_find_directory_without_extra_requests(self):
         assert fs.find("s3://bucket/dir", maxdepth=1, prefix="s") == []
         assert fs._call.call_count == 1
 
+    def test_find_maxdepth_listings_follow_invalidation(self):
+        fs = self._make_fs()
+        self._serve_keys(fs, ("dir/direct", "dir/sub/nested"))
+        assert sorted(fs.find("s3://bucket/dir", maxdepth=2)) == [
+            "bucket/dir/direct",
+            "bucket/dir/sub/nested",
+        ]
+
+        # A write below the subdirectory invalidates its cached listing.
+        self._serve_keys(fs, ("dir/direct", "dir/sub/nested", "dir/sub/new"))
+        fs.invalidate_cache("s3://bucket/dir/sub/new")
+        assert sorted(fs.find("s3://bucket/dir", maxdepth=2)) == [
+            "bucket/dir/direct",
+            "bucket/dir/sub/nested",
+            "bucket/dir/sub/new",
+        ]
+
     def test_find_prefix_counts_levels_from_path(self):
         fs = self._make_fs()
         self._serve_keys(fs, self.FIND_KEYS)
diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py
index 92e36fa6..1f5fc684 100644
--- a/tests/pyathena/filesystem/test_s3_async.py
+++ b/tests/pyathena/filesystem/test_s3_async.py
@@ -1004,17 +1004,6 @@ async def test_find_maxdepth(self, fs):
         result = await fs._find(dir_)
         assert len(result) == 4
 
-        # Each slash in the prefix counts as one level
-        assert await fs._find(dir_, maxdepth=1, prefix="level1/") == []
-        assert await fs._find(dir_, maxdepth=2, prefix="level1/") == [
-            fs._strip_protocol(f"{dir_}/level1/file1.txt")
-        ]
-
-        # An object path returns the object itself
-        assert await fs._find(f"{dir_}/file0.txt", maxdepth=1) == [
-            fs._strip_protocol(f"{dir_}/file0.txt")
-        ]
-
     @pytest.mark.asyncio
     async def test_find_withdirs(self, fs):
         dir_ = (
@@ -1036,8 +1025,6 @@ async def test_find_withdirs(self, fs):
         # Test withdirs=True
         result = await fs._find(dir_, withdirs=True)
         assert len(result) > 4  # Files and directories
-        assert fs._strip_protocol(dir_) in result
-        assert fs._strip_protocol(dir_) in await fs._find(dir_, maxdepth=1, withdirs=True)
 
         # Verify directories are included
         dirs = [r for r in result if not r.endswith(".txt")]

From 089e86e68197df6db5671cc5cfcff1c7294665ae Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 17:29:26 +0900
Subject: [PATCH 05/12] Describe the prefix and fallback rules of find()
 precisely

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

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 246b25f1..837555e2 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -923,7 +923,8 @@ def _find(
             if withdirs:
                 files = [self._directory_object(bucket, key), *files]
         elif key:
-            # As in fsspec, the path itself is returned if it is an object.
+            # As in fsspec, the path itself is returned if it is an object,
+            # or with withdirs if it is a directory.
             try:
                 files = [self.info(path, refresh=refresh)]
             except FileNotFoundError:
@@ -983,8 +984,8 @@ def find(
             **kwargs: Additional arguments including:
                 prefix: Key prefix, relative to the path, to filter the listed keys
                     by. Each slash in the prefix counts as one level of maxdepth.
-                    With withdirs, only the directories whose paths relative to the
-                    path start with the prefix are included.
+                    With withdirs, the directories above the prefix, such as
+                    ``sub`` for ``sub/deep/``, are not included.
                 refresh: If True, bypass the cache and list from S3.
 
         Returns:

From 0d6d6802f17bd7c917345c5e63b0caefdecf6baf Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 17:44:21 +0900
Subject: [PATCH 06/12] Leave the bucket out of find(withdirs=True)

fsspec's recursive copy() keeps the directories that glob() returns
and passes them to cp_file(), which raises ValueError for a bucket, so
copy("s3://bucket/**", ..., recursive=True) failed once the bucket was
included as the root. A bucket path is now not included, as before.

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

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 837555e2..ac55a7ff 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -919,8 +919,9 @@ def _find(
 
         if files:
             # Something is listed below the path, so the path is a directory,
-            # which fsspec includes with the directories.
-            if withdirs:
+            # which fsspec includes with the directories. A bucket is not
+            # included, since cp_file() cannot copy it in a recursive copy.
+            if withdirs and key:
                 files = [self._directory_object(bucket, key), *files]
         elif key:
             # As in fsspec, the path itself is returned if it is an object,
@@ -972,7 +973,8 @@ def find(
         depth limiting and directory inclusion. Uses efficient S3 list operations
         with delimiter handling for performance. As in fsspec, the result
         includes the path itself if withdirs is True and objects exist below
-        it, or if it is an object and nothing is listed below it.
+        it, unless it is a bucket, or if it is an object and nothing is listed
+        below it.
 
         Args:
             path: S3 path to search under (e.g., "s3://bucket/prefix").
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 659e5f2f..95fb1a1c 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -863,6 +863,15 @@ def test_glob_matches_fsspec(self, pattern, maxdepth):
         expected = [p.lstrip("/") for p in memory.glob(f"/bucket/{pattern}", maxdepth=maxdepth)]
         assert sorted(fs.glob(f"s3://bucket/{pattern}", maxdepth=maxdepth)) == expected
 
+    def test_find_withdirs_omits_bucket(self):
+        fs = self._make_fs()
+        self._serve_keys(fs, self.FIND_KEYS)
+
+        # A recursive copy of the expanded paths cannot copy a bucket.
+        assert "bucket" not in fs.find("s3://bucket", withdirs=True)
+        assert "bucket" not in fs.find("s3://bucket", maxdepth=1, withdirs=True)
+        assert "bucket" not in fs.expand_path("s3://bucket/**", recursive=True)
+
     def test_find_directory_without_extra_requests(self):
         fs = self._make_fs()
         self._serve_keys(fs, self.FIND_KEYS)

From 60d71644c602d42ae484c7a3703cab0833d133ce Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 22:57:00 +0900
Subject: [PATCH 07/12] Leave directories out of the conflict check of mv()

With find(withdirs=True) including the path itself, a glob such as
mv("src/**", "src/archive/", recursive=True) pairs the directory src
with the destination src/archive, which is also a source directory
when it exists, so mv() raised ValueError. Directories are not copied,
so _move_paths() now checks only the sources without other sources
below them, without extra requests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py            | 20 ++++++++++++----
 pyathena/filesystem/s3_async.py      |  3 ++-
 tests/pyathena/filesystem/test_s3.py | 36 ++++++++++++++++++++++++++--
 3 files changed, 52 insertions(+), 7 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index ac55a7ff..2532dbd3 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -1488,7 +1488,8 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
         Raises:
             ValueError: If two sources have the same destination, or a
                 destination is another source, which is checked before
-                anything is copied.
+                anything is copied. Directories with other sources below
+                them are not checked, since they are not copied.
         """
         if path1 == path2:
             return
@@ -1517,7 +1518,8 @@ def _move_paths(
 
         Raises:
             ValueError: If two sources have the same destination, or a
-                destination is another source.
+                destination is another source, not counting the directories
+                with other sources below them.
         """
         if isinstance(path1, list) and isinstance(path2, list):
             paths1, paths2 = path1, path2
@@ -1546,10 +1548,20 @@ def _move_paths(
             for p1, p2 in zip(paths1, paths2, strict=False)
             if self._strip_protocol(p1) != self._strip_protocol(p2)
         ]
-        destinations = [self._strip_protocol(p2) for _, p2 in pairs]
+        stripped = [(self._strip_protocol(p1), self._strip_protocol(p2)) for p1, p2 in pairs]
+        # A source with another source below it is a directory, which is
+        # not copied, so only the other sources can conflict.
+        directories: set[str] = set()
+        for source, _ in stripped:
+            parent = source.rpartition("/")[0]
+            while parent and parent not in directories:
+                directories.add(parent)
+                parent = parent.rpartition("/")[0]
+        files = [(source, dest) for source, dest in stripped if source not in directories]
+        destinations = [dest for _, dest in files]
         if len(set(destinations)) != len(destinations):
             raise ValueError("Cannot move several paths to the same destination.")
-        if {self._strip_protocol(p1) for p1, _ in pairs}.intersection(destinations):
+        if {source for source, _ in files}.intersection(destinations):
             raise ValueError("Cannot move a path onto another path that is moved.")
         return pairs
 
diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py
index a7d933bc..cfd041a5 100644
--- a/pyathena/filesystem/s3_async.py
+++ b/pyathena/filesystem/s3_async.py
@@ -355,7 +355,8 @@ async def _mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> N
         Raises:
             ValueError: If two sources have the same destination, or a
                 destination is another source, which is checked before
-                anything is copied.
+                anything is copied. Directories with other sources below
+                them are not checked, since they are not copied.
         """
         if path1 == path2:
             return
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 95fb1a1c..25db7451 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -764,13 +764,23 @@ def test_find_refresh_bypasses_cached_listings(self):
 
     @staticmethod
     def _serve_keys(fs, keys):
-        # Answer the ListObjectsV2 and HeadObject requests of "bucket" from
-        # the given keys.
+        # Answer the ListObjectsV2, HeadObject, CopyObject, and DeleteObjects
+        # requests of "bucket" from a set of the given keys, which is returned.
+        keys = set(keys)
+
         def call(method, **kwargs):
             if method is fs._client.head_object:
                 if kwargs["Key"] not in keys:
                     raise FileNotFoundError(kwargs["Key"])
                 return {"ContentLength": 0}
+            if method is fs._client.copy_object:
+                if kwargs["CopySource"]["Key"] not in keys:
+                    raise FileNotFoundError(kwargs["CopySource"]["Key"])
+                keys.add(kwargs["Key"])
+                return {}
+            if method is fs._client.delete_objects:
+                keys.difference_update(o["Key"] for o in kwargs["Delete"]["Objects"])
+                return {}
             prefix, delimiter = kwargs["Prefix"], kwargs["Delimiter"]
             contents, prefixes = [], set()
             for key in sorted(keys):
@@ -788,6 +798,7 @@ def call(method, **kwargs):
             }
 
         fs._call.side_effect = call
+        return keys
 
     @staticmethod
     def _memory_fs(keys):
@@ -1538,6 +1549,27 @@ def test_mv_keeps_given_paths(self):
         fs._copy_file.assert_called_once_with("s3://bucket/src", "s3://bucket/dst/")
         fs._delete_objects.assert_called_once_with(["s3://bucket/src"])
 
+    @pytest.mark.parametrize(
+        ("keys", "path1", "path2", "expected"),
+        [
+            # The directory itself, which find() includes, is not copied.
+            ({"d/a", "d/b"}, "s3://bucket/d/**", "s3://bucket/out/", {"out/a", "out/b"}),
+            # The directory moved onto an existing subdirectory is no conflict.
+            (
+                {"src/a", "src/archive/x"},
+                "s3://bucket/src/**",
+                "s3://bucket/src/archive/",
+                {"src/archive/a", "src/archive/archive/x"},
+            ),
+        ],
+    )
+    def test_mv_glob_with_directories(self, keys, path1, path2, expected):
+        fs = self._make_fs()
+        store = self._serve_keys(fs, keys)
+
+        fs.mv(path1, path2, recursive=True)
+        assert store == expected
+
     def test_mv_nothing_within_maxdepth(self):
         # Only directories within maxdepth: nothing is moved, as with copy().
         fs = self._make_fs()

From 8e4be2a85fbd95fd94571002fbd504a02b7e04dc Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 22:59:28 +0900
Subject: [PATCH 08/12] Skip only directory-to-directory pairs in the mv()
 conflict check

An object with keys below it counts as a directory, but it is copied,
so excluding every directory source let its copy overwrite another
source. Only a directory moved onto another directory is skipped now.

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

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 2532dbd3..36042656 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -1488,8 +1488,8 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
         Raises:
             ValueError: If two sources have the same destination, or a
                 destination is another source, which is checked before
-                anything is copied. Directories with other sources below
-                them are not checked, since they are not copied.
+                anything is copied. A directory, which has other sources
+                below it and is not copied, may move onto another directory.
         """
         if path1 == path2:
             return
@@ -1518,8 +1518,8 @@ def _move_paths(
 
         Raises:
             ValueError: If two sources have the same destination, or a
-                destination is another source, not counting the directories
-                with other sources below them.
+                destination is another source, except a directory, which has
+                other sources below it, moved onto another directory.
         """
         if isinstance(path1, list) and isinstance(path2, list):
             paths1, paths2 = path1, path2
@@ -1550,18 +1550,19 @@ def _move_paths(
         ]
         stripped = [(self._strip_protocol(p1), self._strip_protocol(p2)) for p1, p2 in pairs]
         # A source with another source below it is a directory, which is
-        # not copied, so only the other sources can conflict.
+        # not copied, so a directory moved onto another one is no conflict.
         directories: set[str] = set()
         for source, _ in stripped:
             parent = source.rpartition("/")[0]
             while parent and parent not in directories:
                 directories.add(parent)
                 parent = parent.rpartition("/")[0]
-        files = [(source, dest) for source, dest in stripped if source not in directories]
-        destinations = [dest for _, dest in files]
+        destinations = [
+            dest for source, dest in stripped if not (source in directories and dest in directories)
+        ]
         if len(set(destinations)) != len(destinations):
             raise ValueError("Cannot move several paths to the same destination.")
-        if {source for source, _ in files}.intersection(destinations):
+        if {source for source, _ in stripped}.intersection(destinations):
             raise ValueError("Cannot move a path onto another path that is moved.")
         return pairs
 
diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py
index cfd041a5..2b0a97d1 100644
--- a/pyathena/filesystem/s3_async.py
+++ b/pyathena/filesystem/s3_async.py
@@ -355,8 +355,8 @@ async def _mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> N
         Raises:
             ValueError: If two sources have the same destination, or a
                 destination is another source, which is checked before
-                anything is copied. Directories with other sources below
-                them are not checked, since they are not copied.
+                anything is copied. A directory, which has other sources
+                below it and is not copied, may move onto another directory.
         """
         if path1 == path2:
             return
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 25db7451..38ed47a7 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -1517,6 +1517,11 @@ def test_cp_file_directory(self):
             (["s3://bucket/a", "s3://bucket/b"], ["s3://bucket/c", "s3a://bucket/c"]),
             # A destination that is another source.
             (["s3://bucket/a", "s3://bucket/b"], ["s3://bucket/b", "s3://bucket/a"]),
+            # An object with keys below it is checked like a file.
+            (
+                ["s3://bucket/d", "s3://bucket/d/x", "s3://bucket/e"],
+                ["s3://bucket/e", "s3://bucket/o/x", "s3://bucket/o/e"],
+            ),
         ],
     )
     def test_mv_conflicting_destinations(self, path1, path2):

From e37e04642628ed42f9c3f9ffe97c94c05d3785d7 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 23:10:35 +0900
Subject: [PATCH 09/12] Confirm with HeadObject that a directory in mv() is not
 an object

A source with other sources below it can still be an object, which
_copy_file() copies, so treating every such source as a directory let
mv() overwrite another source. A source that would conflict is now
left out only when HeadObject finds no object at its key, which costs
one request only for those sources.

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

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 36042656..bb77d21a 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -10,6 +10,7 @@
 import os.path
 import re
 import time
+from collections import Counter
 from collections.abc import Callable, Iterator, Mapping
 from concurrent.futures import Future, as_completed, wait
 from copy import deepcopy
@@ -1488,8 +1489,8 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
         Raises:
             ValueError: If two sources have the same destination, or a
                 destination is another source, which is checked before
-                anything is copied. A directory, which has other sources
-                below it and is not copied, may move onto another directory.
+                anything is copied. A directory with no object at its key,
+                which is not copied, does not conflict.
         """
         if path1 == path2:
             return
@@ -1518,8 +1519,8 @@ def _move_paths(
 
         Raises:
             ValueError: If two sources have the same destination, or a
-                destination is another source, except a directory, which has
-                other sources below it, moved onto another directory.
+                destination is another source, except for a directory with
+                no object at its key, which is not copied.
         """
         if isinstance(path1, list) and isinstance(path2, list):
             paths1, paths2 = path1, path2
@@ -1549,20 +1550,24 @@ def _move_paths(
             if self._strip_protocol(p1) != self._strip_protocol(p2)
         ]
         stripped = [(self._strip_protocol(p1), self._strip_protocol(p2)) for p1, p2 in pairs]
-        # A source with another source below it is a directory, which is
-        # not copied, so a directory moved onto another one is no conflict.
+        sources = {source for source, _ in stripped}
+        counts = Counter(dest for _, dest in stripped)
+        # A source with another source below it may be a directory.
         directories: set[str] = set()
         for source, _ in stripped:
             parent = source.rpartition("/")[0]
             while parent and parent not in directories:
                 directories.add(parent)
                 parent = parent.rpartition("/")[0]
-        destinations = [
-            dest for source, dest in stripped if not (source in directories and dest in directories)
-        ]
-        if len(set(destinations)) != len(destinations):
-            raise ValueError("Cannot move several paths to the same destination.")
-        if {source for source, _ in stripped}.intersection(destinations):
+        for source, dest in stripped:
+            if counts[dest] == 1 and dest not in sources:
+                continue
+            # A directory without an object at its key is not copied, so it
+            # writes no destination.
+            if source in directories and self._head_object(source) is None:
+                continue
+            if counts[dest] > 1:
+                raise ValueError("Cannot move several paths to the same destination.")
             raise ValueError("Cannot move a path onto another path that is moved.")
         return pairs
 
diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py
index 2b0a97d1..53d08a6f 100644
--- a/pyathena/filesystem/s3_async.py
+++ b/pyathena/filesystem/s3_async.py
@@ -355,8 +355,8 @@ async def _mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> N
         Raises:
             ValueError: If two sources have the same destination, or a
                 destination is another source, which is checked before
-                anything is copied. A directory, which has other sources
-                below it and is not copied, may move onto another directory.
+                anything is copied. A directory with no object at its key,
+                which is not copied, does not conflict.
         """
         if path1 == path2:
             return
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 38ed47a7..1819a14d 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -1517,11 +1517,6 @@ def test_cp_file_directory(self):
             (["s3://bucket/a", "s3://bucket/b"], ["s3://bucket/c", "s3a://bucket/c"]),
             # A destination that is another source.
             (["s3://bucket/a", "s3://bucket/b"], ["s3://bucket/b", "s3://bucket/a"]),
-            # An object with keys below it is checked like a file.
-            (
-                ["s3://bucket/d", "s3://bucket/d/x", "s3://bucket/e"],
-                ["s3://bucket/e", "s3://bucket/o/x", "s3://bucket/o/e"],
-            ),
         ],
     )
     def test_mv_conflicting_destinations(self, path1, path2):
@@ -1575,6 +1570,29 @@ def test_mv_glob_with_directories(self, keys, path1, path2, expected):
         fs.mv(path1, path2, recursive=True)
         assert store == expected
 
+    @pytest.mark.parametrize(
+        ("keys", "path2"),
+        [
+            # Onto another source that is only an object.
+            ({"d", "d/x", "e"}, ["e", "o/x", "o/e"]),
+            # Onto another source that also has keys below it.
+            ({"d", "d/x", "d/x/y"}, ["d/x", "o", "o/y"]),
+        ],
+    )
+    def test_mv_objects_with_keys_below_conflict(self, keys, path2):
+        # An object that also has keys below it is copied, so moving it onto
+        # another source raises before anything is copied.
+        fs = self._make_fs()
+        store = self._serve_keys(fs, keys)
+
+        with pytest.raises(ValueError, match="another path that is moved"):
+            fs.mv(
+                [f"s3://bucket/{k}" for k in sorted(keys)],
+                [f"s3://bucket/{k}" for k in path2],
+                recursive=True,
+            )
+        assert store == keys
+
     def test_mv_nothing_within_maxdepth(self):
         # Only directories within maxdepth: nothing is moved, as with copy().
         fs = self._make_fs()

From bf24179b7e5f310f6ffe2697bfea7e31dc20f981 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sat, 3 Oct 2026 23:24:28 +0900
Subject: [PATCH 10/12] Leave skipped directories out of the duplicate
 destinations in mv()

A directory without an object at its key writes nothing, but it still
counted toward the duplicate destinations, so a directory sharing its
destination with an object made mv() raise. The skipped directories
are now dropped before both checks.

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

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index bb77d21a..479744ed 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -1559,16 +1559,23 @@ def _move_paths(
             while parent and parent not in directories:
                 directories.add(parent)
                 parent = parent.rpartition("/")[0]
-        for source, dest in stripped:
-            if counts[dest] == 1 and dest not in sources:
-                continue
-            # A directory without an object at its key is not copied, so it
-            # writes no destination.
-            if source in directories and self._head_object(source) is None:
-                continue
+        # A directory without an object at its key is not copied, so it
+        # writes no destination and is left out of the checks.
+        writers = [
+            (source, dest)
+            for source, dest in stripped
+            if not (
+                (counts[dest] > 1 or dest in sources)
+                and source in directories
+                and self._head_object(source) is None
+            )
+        ]
+        counts = Counter(dest for _, dest in writers)
+        for _, dest in writers:
             if counts[dest] > 1:
                 raise ValueError("Cannot move several paths to the same destination.")
-            raise ValueError("Cannot move a path onto another path that is moved.")
+            if dest in sources:
+                raise ValueError("Cannot move a path onto another path that is moved.")
         return pairs
 
     def cp_file(
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 1819a14d..90035200 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -1561,6 +1561,13 @@ def test_mv_keeps_given_paths(self):
                 "s3://bucket/src/archive/",
                 {"src/archive/a", "src/archive/archive/x"},
             ),
+            # A directory shares its destination with an object.
+            (
+                {"d/x", "e/y"},
+                ["s3://bucket/d", "s3://bucket/d/x", "s3://bucket/e/y"],
+                ["s3://bucket/e", "s3://bucket/e", "s3://bucket/out"],
+                {"e", "out"},
+            ),
         ],
     )
     def test_mv_glob_with_directories(self, keys, path1, path2, expected):
@@ -1592,6 +1599,9 @@ def test_mv_objects_with_keys_below_conflict(self, keys, path2):
                 recursive=True,
             )
         assert store == keys
+        methods = {c.args[0] for c in fs._call.call_args_list}
+        assert fs._client.copy_object not in methods
+        assert fs._client.delete_objects not in methods
 
     def test_mv_nothing_within_maxdepth(self):
         # Only directories within maxdepth: nothing is moved, as with copy().

From 68cec03d821ba417576d2c215c5705cd994690fb Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sun, 4 Oct 2026 00:33:08 +0900
Subject: [PATCH 11/12] Protect sources left in place and null versions in mv()

_move_paths() dropped the sources whose destination is the source
itself before the checks, so mv([a, b], [b, b]) copied a over b, which
was meant to stay in place. It also compared version-qualified paths as
text, but a write to a key replaces its "null" version, so
mv([a, b?versionId=null], [b, out]) overwrote b before copying it, and
mv([b?versionId=null], [b]) copied b onto itself and then deleted it.

The paths are now compared by what they name, with the "null" version
as the key itself, and the sources left in place count as sources.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyathena/filesystem/s3.py            | 53 ++++++++++++++++++++--------
 pyathena/filesystem/s3_async.py      |  6 ++--
 tests/pyathena/filesystem/test_s3.py | 26 ++++++++++++++
 3 files changed, 67 insertions(+), 18 deletions(-)

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 479744ed..9de7ec6a 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -1475,8 +1475,9 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
         ``AbstractFileSystem.mv()`` instead removes ``path1`` by expanding it
         again, which also deletes copies placed where ``path1`` matches them
         and files that ``maxdepth`` kept from being copied. A file whose
-        destination is the file itself is left in place, and directories,
-        which S3 does not store as objects, are not copied.
+        destination is the file itself, or the ``null`` version of a file
+        moved to the file, is left in place, and directories, which S3 does
+        not store as objects, are not copied.
 
         Args:
             path1: Source S3 path, glob pattern, or list of paths.
@@ -1488,9 +1489,9 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
 
         Raises:
             ValueError: If two sources have the same destination, or a
-                destination is another source, which is checked before
-                anything is copied. A directory with no object at its key,
-                which is not copied, does not conflict.
+                destination is another source, including one left in place,
+                which is checked before anything is copied. A directory with
+                no object at its key, which is not copied, does not conflict.
         """
         if path1 == path2:
             return
@@ -1515,12 +1516,14 @@ def _move_paths(
 
         Returns:
             The source and destination paths, except the sources whose
-            destination is the source itself.
+            destination is the source itself or, for a ``null`` version, the
+            key of the source.
 
         Raises:
             ValueError: If two sources have the same destination, or a
-                destination is another source, except for a directory with
-                no object at its key, which is not copied.
+                destination is another source, including one left in place,
+                except for a directory with no object at its key, which is not
+                copied.
         """
         if isinstance(path1, list) and isinstance(path2, list):
             paths1, paths2 = path1, path2
@@ -1543,18 +1546,19 @@ def _move_paths(
                 )
             )
             paths2 = other_paths(paths1, path2, exists=exists, flatten=not source_is_str)
-        # The paths are copied as given, and compared without the protocol.
-        pairs = [
-            (p1, p2)
+        # The paths are copied as given, and compared by what they name.
+        named = [
+            (p1, p2, self._move_target(p1), self._move_target(p2))
             for p1, p2 in zip(paths1, paths2, strict=False)
-            if self._strip_protocol(p1) != self._strip_protocol(p2)
         ]
-        stripped = [(self._strip_protocol(p1), self._strip_protocol(p2)) for p1, p2 in pairs]
-        sources = {source for source, _ in stripped}
+        pairs = [(p1, p2) for p1, p2, source, dest in named if source != dest]
+        stripped = [(source, dest) for _, _, source, dest in named if source != dest]
+        # The sources left in place count too; a copy onto one overwrites it.
+        sources = {source for _, _, source, _ in named}
         counts = Counter(dest for _, dest in stripped)
         # A source with another source below it may be a directory.
         directories: set[str] = set()
-        for source, _ in stripped:
+        for source in sources:
             parent = source.rpartition("/")[0]
             while parent and parent not in directories:
                 directories.add(parent)
@@ -1578,6 +1582,25 @@ def _move_paths(
                 raise ValueError("Cannot move a path onto another path that is moved.")
         return pairs
 
+    def _move_target(self, path: str) -> str:
+        """Return what a path of a move names, for comparing the paths.
+
+        A write to a key replaces its ``null`` version, which the objects of a
+        bucket without versioning have, so that version names the key itself.
+
+        Args:
+            path: S3 path, possibly with a version ID.
+
+        Returns:
+            The path in ``bucket/key`` form, with the version ID unless it is
+            ``null``.
+        """
+        bucket, key, version_id = self.parse_path(path)
+        target = f"{bucket}/{key}" if key else bucket
+        if version_id and version_id != "null":
+            return f"{target}?versionId={version_id}"
+        return target
+
     def cp_file(
         self, path1: str, path2: str, recursive=False, maxdepth=None, on_error=None, **kwargs
     ):
diff --git a/pyathena/filesystem/s3_async.py b/pyathena/filesystem/s3_async.py
index 53d08a6f..16269606 100644
--- a/pyathena/filesystem/s3_async.py
+++ b/pyathena/filesystem/s3_async.py
@@ -354,9 +354,9 @@ async def _mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> N
 
         Raises:
             ValueError: If two sources have the same destination, or a
-                destination is another source, which is checked before
-                anything is copied. A directory with no object at its key,
-                which is not copied, does not conflict.
+                destination is another source, including one left in place,
+                which is checked before anything is copied. A directory with
+                no object at its key, which is not copied, does not conflict.
         """
         if path1 == path2:
             return
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index 90035200..f50050a9 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -1517,6 +1517,13 @@ def test_cp_file_directory(self):
             (["s3://bucket/a", "s3://bucket/b"], ["s3://bucket/c", "s3a://bucket/c"]),
             # A destination that is another source.
             (["s3://bucket/a", "s3://bucket/b"], ["s3://bucket/b", "s3://bucket/a"]),
+            # A destination that is a source left in place.
+            (["s3://bucket/a", "s3://bucket/b"], ["s3://bucket/b", "s3://bucket/b"]),
+            # A destination whose "null" version is another source.
+            (
+                ["s3://bucket/a", "s3://bucket/b?versionId=null"],
+                ["s3://bucket/b", "s3://bucket/out"],
+            ),
         ],
     )
     def test_mv_conflicting_destinations(self, path1, path2):
@@ -1603,6 +1610,25 @@ def test_mv_objects_with_keys_below_conflict(self, keys, path2):
         assert fs._client.copy_object not in methods
         assert fs._client.delete_objects not in methods
 
+    def test_mv_versions_onto_their_key(self):
+        fs = self._make_fs()
+        self._serve_keys(fs, {"b"})
+
+        # The "null" version is the object at the key, so it stays in place.
+        fs.mv(["s3://bucket/b?versionId=null"], ["s3://bucket/b"])
+        fs._call.assert_not_called()
+
+        # Another version is copied onto the key, and then deleted.
+        fs.mv(["s3://bucket/b?versionId=v1"], ["s3://bucket/b"])
+        copies = [c.kwargs for c in fs._call.call_args_list if c.args[0] is fs._client.copy_object]
+        assert [(c["CopySource"].get("VersionId"), c["Key"]) for c in copies] == [("v1", "b")]
+        deletes = [
+            c.kwargs["Delete"]["Objects"]
+            for c in fs._call.call_args_list
+            if c.args[0] is fs._client.delete_objects
+        ]
+        assert deletes == [[{"Key": "b", "VersionId": "v1"}]]
+
     def test_mv_nothing_within_maxdepth(self):
         # Only directories within maxdepth: nothing is moved, as with copy().
         fs = self._make_fs()

From ace8248c44e166e8dd22a07431863a0e37edc497 Mon Sep 17 00:00:00 2001
From: laughingman7743 <laughingman7743@gmail.com>
Date: Sun, 4 Oct 2026 00:42:10 +0900
Subject: [PATCH 12/12] Keep versioned sources out of the directory exemption
 in mv()

The exemption looked up the source without its version, since the
"null" version is compared as the key. A version hidden by a delete
marker then looked like a directory when keys existed below its key,
so its copy skipped the conflict check. A path with a version always
names an object, so it is no longer exempted.

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

diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py
index 9de7ec6a..69f8ed98 100644
--- a/pyathena/filesystem/s3.py
+++ b/pyathena/filesystem/s3.py
@@ -1552,10 +1552,10 @@ def _move_paths(
             for p1, p2 in zip(paths1, paths2, strict=False)
         ]
         pairs = [(p1, p2) for p1, p2, source, dest in named if source != dest]
-        stripped = [(source, dest) for _, _, source, dest in named if source != dest]
+        moved = [(p1, source, dest) for p1, _, source, dest in named if source != dest]
         # The sources left in place count too; a copy onto one overwrites it.
         sources = {source for _, _, source, _ in named}
-        counts = Counter(dest for _, dest in stripped)
+        counts = Counter(dest for _, _, dest in moved)
         # A source with another source below it may be a directory.
         directories: set[str] = set()
         for source in sources:
@@ -1564,13 +1564,15 @@ def _move_paths(
                 directories.add(parent)
                 parent = parent.rpartition("/")[0]
         # A directory without an object at its key is not copied, so it
-        # writes no destination and is left out of the checks.
+        # writes no destination and is left out of the checks. A path with a
+        # version always names an object.
         writers = [
             (source, dest)
-            for source, dest in stripped
+            for p1, source, dest in moved
             if not (
                 (counts[dest] > 1 or dest in sources)
                 and source in directories
+                and not self.parse_path(p1)[2]
                 and self._head_object(source) is None
             )
         ]
diff --git a/tests/pyathena/filesystem/test_s3.py b/tests/pyathena/filesystem/test_s3.py
index f50050a9..2f2dc87e 100644
--- a/tests/pyathena/filesystem/test_s3.py
+++ b/tests/pyathena/filesystem/test_s3.py
@@ -1610,6 +1610,21 @@ def test_mv_objects_with_keys_below_conflict(self, keys, path2):
         assert fs._client.copy_object not in methods
         assert fs._client.delete_objects not in methods
 
+    def test_mv_version_with_keys_below_conflicts(self):
+        # A version names an object even with keys below its key, so it is
+        # not taken for a directory when no current object exists at the key.
+        fs = self._make_fs()
+        self._serve_keys(fs, {"d/x", "a"})
+
+        with pytest.raises(ValueError, match="same destination"):
+            fs.mv(
+                ["s3://bucket/d?versionId=null", "s3://bucket/d/x", "s3://bucket/a"],
+                ["s3://bucket/out", "s3://bucket/x", "s3://bucket/out"],
+            )
+        methods = {c.args[0] for c in fs._call.call_args_list}
+        assert fs._client.copy_object not in methods
+        assert fs._client.delete_objects not in methods
+
     def test_mv_versions_onto_their_key(self):
         fs = self._make_fs()
         self._serve_keys(fs, {"b"})