Model S3 paths with one value class instead of parsing strings at each call site #1049

Use case

pyathena/filesystem/ handles S3 paths as strings, and each call site takes them apart again in its own way. Line numbers are from pyathena/filesystem/s3.py on master 28ec68d.

• Scheme. Two parsers remove it: the PATTERN_PATH regular expression in parse_path() (s3://, s3a:// or none) and fsspec's _strip_protocol() (10 calls). get_file() runs both (parse_path(self._strip_protocol(rpath)), L2193).
• Bucket and key. parse_path() splits them with the regular expression and keeps a trailing slash in the key, while _strip_protocol() removes it. Other call sites decide "bucket or key path" themselves: ls() checks "/" in path (L635), rm() treats bucket// as the bucket with not key or not key.strip("/") (L1124), and info() uses key.rstrip("/") for directory entries (L797, L829).
• Version. parse_path() is called 30 times in s3.py and 3 times in s3_async.py, each unpacking a (bucket, key, version_id) tuple. 7 places build a version-qualified path by hand with f"...?versionId={version_id}" (_head_object(), the paths of _delete_objects, _move_target(), S3File.__init__(), ...), and _head_object() and invalidate_cache() remove the version by splitting at the first ? (path.partition("?")). _move_target() adds its own rule for the null version.
• Display. Log and error messages rebuild s3://{bucket}/{key}?versionId={version_id} by hand.

The string handling comes from the s3fs design that this filesystem started from. Changing how a version is parsed or spelled means finding every one of these places by hand: #979 changes the parsing rule (only a trailing ?versionId= query is a version, so keys may contain ?), and the partition("?") sites are the ones that change would silently miss.

Proposed change

Add a public frozen value class for an S3 path, in a new module pyathena/filesystem/s3_path.py, following ExecuteOptions (pyathena/options.py) for the frozen dataclass style, and make it the one place that knows the path syntax:

@dataclass(frozen=True)
class S3Path:
    bucket: str
    key: str | None = None  # None for a bucket path; a trailing "/" is kept ("dir/" names a directory, #998)
    version_id: str | None = None

    @classmethod
    def parse(cls, path: str | os.PathLike[str]) -> S3Path: ...
        # s3://, s3a:// or no scheme; bucket; key; trailing version query. ValueError otherwise.

    @property
    def is_bucket(self) -> bool: ...  # no key, or a key of only slashes ("bucket//")

    @property
    def name(self) -> str: ...  # "bucket/key" without the trailing slash, or "bucket": what _strip_protocol() gives, used as the fsspec name and cache key

    @property
    def uri(self) -> str: ...  # "s3://bucket/key?versionId=<id>", for messages

    def __str__(self) -> str: ...  # name, plus "?versionId=<id>" when there is a version

    def with_version_id(self, version_id: str | None) -> S3Path: ...

• Use S3Path at the internal call sites of parse_path(), in place of _strip_protocol() followed by parsing, of the bucket/key checks above, and of the hand-built and hand-split version queries and URIs.
• Keep S3FileSystem.parse_path() and AioS3FileSystem.parse_path() as public methods returning the same tuple, implemented with S3Path.parse(), and keep S3FileSystem.PATTERN_PATH.
• Keep fsspec's _strip_protocol() and _parent() for the names fsspec itself builds (glob, find, expand_path); S3Path.name returns the same string, so cache keys do not change.
• No behavior change: the parsing rule and the trailing-slash semantics stay as they are, and Keys containing '?' are rejected, and pinned or listed versions are not addressable #979 changes the version rule afterward in one place.

Decided with the maintainer: the root path ("", "/", "s3://") stays outside S3Path. parse() rejects it as parse_path() does today, so bucket is never empty, and ls()/info()/invalidate_cache() keep checking the root before parsing. parse() accepts the schemes that PATTERN_PATH accepts today (s3://, s3a://, none).

Out of scope: S3 access point ARNs and other bucket name forms that PATTERN_PATH does not accept, and any change to fsspec's naming or cache key conventions.

Alternatives considered: small private helpers (strip or append a version query) next to parse_path(). They would remove the duplicated version handling, but leave the scheme and bucket/key handling scattered and give the path no type.

Validation plan (if implementing)

• Unit tests for S3Path (parse with each scheme, bucket paths including bucket/ and bucket//, trailing-slash keys, versions in each spelling, name, uri, str(), with_version_id(), invalid paths), next to the existing test_parse_path cases.
• The existing offline filesystem tests and the S3 integration tests in tests/pyathena/filesystem/ pass unchanged.