Recursive get(), copy() and mv() mishandle directory entries

[laughingman7743]

Problem

S3FileSystem.get_file() and cp_file() do not handle the directory entries that fsspec passes to them during recursive operations, so recursive get() and mv() fail.

1. get_file() does not follow the fsspec get_file() contract, so recursive get() fails. pyathena/filesystem/s3.py:1475-1499 only checks whether lpath is an existing local directory and then opens lpath for writing.
   • A directory rpath is written as an empty local file instead of creating a local directory. Recursive get("s3://bucket/src", "out", recursive=True) first writes out as an empty file and then raises NotADirectoryError for out/a.
   • Missing parent directories of lpath are not created: get_file("s3://bucket/src/a", "new/a") raises FileNotFoundError.
   • A file-like lpath raises TypeError: stat: path should be string, bytes, os.PathLike or integer, not BytesIO.
   • outfile is documented as unused and ignored: get_file(rpath, "x", outfile=buf) writes to the local file x and leaves buf empty.
   • AioS3FileSystem._get_file() delegates to the sync get_file() (pyathena/filesystem/s3_async.py:169-170), so the async filesystem has the same behavior.
2. cp_file() copies directory entries as objects, so mv(recursive=True) fails. fsspec's recursive copy() passes the source directory and its subdirectories to cp_file(). cp_file() sends CopyObject for them (s3.py:1173-1194; s3_async.py:232-266), which fails with NoSuchKey because no such object exists.
   • copy(recursive=True) still succeeds, because fsspec ignores FileNotFoundError in recursive copies, but every directory entry costs a failed CopyObject.
   • mv(recursive=True) calls copy(..., on_error="raise"), so it raises FileNotFoundError. With S3FileSystem, nothing is copied or deleted. With AioS3FileSystem, the copies run concurrently, so every file is copied to the destination and the source is not deleted, which leaves both trees.

Expected:

• get_file() behaves like fsspec's AbstractFileSystem.get_file(): create a local directory for a directory rpath, create missing parent directories, and write to a file-like lpath or to outfile;
• cp_file() skips directory entries, so mv(recursive=True) moves the tree.

Reproduction

import io, os, tempfile
from unittest.mock import MagicMock
from pyathena.filesystem.s3 import S3FileSystem
from pyathena.filesystem.s3_async import AioS3FileSystem

def make_call(objects):
    def call(method, Bucket, Key=None, Prefix="", Delimiter=None, CopySource=None, **kwargs):
        name = method._extract_mock_name().split(".")[-1]
        if name == "list_objects_v2":
            rest = {k[len(Prefix):] for k in objects if k.startswith(Prefix)}
            return {"Contents": [{"Key": Prefix + r, "Size": 1} for r in rest if not (Delimiter and "/" in r)],
                    "CommonPrefixes": [{"Prefix": Prefix + d + "/"}
                                       for d in {r.split("/")[0] for r in rest if Delimiter and "/" in r}]}
        if name == "delete_objects":
            for o in kwargs["Delete"]["Objects"]:
                objects.pop(o["Key"], None)
            return {}
        source = CopySource["Key"] if CopySource else Key
        if source not in objects:
            raise FileNotFoundError(f"NoSuchKey: {source}")
        if name == "copy_object":
            objects[Key] = objects[source]
            return {}
        return {"ContentLength": 1, "ETag": '"e"', "Body": io.BytesIO(objects[Key])}
    return call

fs = S3FileSystem(connection=MagicMock(), skip_instance_cache=True)
fs._call = make_call({"src/a": b"A", "src/sub/b": b"B"})
with tempfile.TemporaryDirectory() as tmp:
    try:
        fs.get("s3://bucket/src", f"{tmp}/out", recursive=True)
    except Exception as e:
        print(repr(e), "out is a file:", os.path.isfile(f"{tmp}/out"))
    # NotADirectoryError(20, 'Not a directory') out is a file: True
    try:
        fs.get_file("s3://bucket/src/a", f"{tmp}/new/a")
    except Exception as e:
        print(repr(e))
    # FileNotFoundError(2, 'No such file or directory')
    buf = io.BytesIO()
    fs.get_file("s3://bucket/src/a", f"{tmp}/x", outfile=buf)
    print(buf.getvalue(), os.path.isfile(f"{tmp}/x"))
    # b'' True
try:
    fs.get_file("s3://bucket/src/a", io.BytesIO())
except Exception as e:
    print(repr(e))
# TypeError('stat: path should be string, bytes, os.PathLike or integer, not BytesIO')

for cls in (S3FileSystem, AioS3FileSystem):
    objects = {"src/a": b"A", "src/sub/b": b"B"}
    fs = cls(connection=MagicMock(), skip_instance_cache=True)
    (fs._sync_fs if cls is AioS3FileSystem else fs)._call = make_call(objects)
    try:
        fs.mv("s3://bucket/src", "s3://bucket/dst", recursive=True)
    except Exception as e:
        print(cls.__name__, repr(e), sorted(objects))
# S3FileSystem FileNotFoundError('NoSuchKey: src') ['src/a', 'src/sub/b']
# AioS3FileSystem FileNotFoundError('NoSuchKey: src') ['dst/a', 'dst/sub/b', 'src/a', 'src/sub/b']

Environment

• PyAthena master e0e85da, fsspec 2026.9.0, botocore from uv.lock, Python 3.13.1.
• Found in a full audit of pyathena/filesystem/. Unless noted, checked offline with mocked S3 responses (fs._call replaced) or botocore's Stubber with dummy credentials.

Proposed fix (optional)

• Rewrite S3FileSystem.get_file() after fsspec's AbstractFileSystem.get_file(): honor a file-like lpath and outfile, call os.makedirs(lpath, exist_ok=True) for a directory rpath, and create the parent directory of lpath before opening it.
• In cp_file() and AioS3FileSystem._cp_file(), return without copying when the source info is a directory.
• Add offline tests for recursive get() and for mv(recursive=True) on both filesystems.

[laughingman7743]

Split: PR #990 now fixes item 1 (get_file()) only. Item 2 (cp_file() sending CopyObject for directory entries) moved to #1008. Skipping directories in cp_file() would let fsspec's copy-then-remove mv() delete overlapping copies, so the mv() guard needs its own decision there.