Fix null-version moves in versioning-enabled S3 buckets

docs/filesystem.md

@@ -261,6 +261,23 @@ for version in versions:
 Version-aware operations require the `s3:GetObjectVersion` and
 `s3:ListBucketVersions` permissions.
 
+Moving a `?versionId=null` source onto its own key copies that version and
+deletes the source version when bucket versioning is enabled. If versioning has
+never been enabled or is suspended, the move leaves the source in place: copying
+onto the key would replace the null version that the subsequent deletion removes.
+The same distinction applies to move conflict checks.
+
+A move compares a null version with its unversioned key using `GetBucketVersioning`,
+which requires `s3:GetBucketVersioning`. Each relevant bucket is looked up once
+during planning; other moves do not make this request. The result is not cached
+between moves. A failed lookup stops the move before any copy or deletion.
+Directory buckets do not support versioning, so their null versions are treated
+as the key itself without a versioning lookup.
+
+The opt-in S3 versioning integration tests create dedicated temporary buckets and
+check unversioned, enabled, and suspended states. See [Testing](testing.md) for the
+command and required test permissions.
+
 ## Bucket lifecycle
 
 Bucket creation and deletion are infrastructure-level changes and are disabled by
@@ -387,6 +404,13 @@ paths, and pass the results to `copy_pairs()` and `move_pairs()`. A `sources`,
 `destination_is_dir` or `missing` that a rule needs and that is not passed raises
 `ValueError`.
 
+For moves that compare a null version with its unversioned key, the caller also
+looks up bucket versioning and passes the names of the versioning-enabled buckets
+as `versioning_enabled_buckets` to both `conflict_candidates()` and `move_pairs()`.
+Pass a collection of bucket names, such as a set, rather than a single string.
+The default empty collection treats null versions as their keys, as in unversioned
+or suspended buckets. The model makes no AWS requests.
+
 ```python
 from pyathena.filesystem.s3_path_pairing import S3PathPairing
 

docs/testing.md

@@ -158,6 +158,37 @@ The offline-only `.env` example disables EC2 metadata to prevent implicit creden
 `--noconftest` excludes the AWS session hooks and fixtures; disabling the rerun plugin also avoids its local socket setup in restricted environments.
 Use this invocation only for self-contained modules; integration tests need their normal fixtures and a real AWS environment.
 
+### S3 null-version moves
+
+The null-version move regression tests create three dedicated temporary buckets:
+unversioned, versioning-enabled, and versioning-suspended. They run only when
+explicitly enabled, because provisioning requires additional permissions and a
+15-minute wait after first enabling versioning. Run them serially:
+
+```bash
+AWS_ATHENA_S3_VERSIONING_TESTS=1 uv run --env-file .env pytest -n 1 \
+  tests/pyathena/filesystem/test_s3.py \
+  tests/pyathena/filesystem/test_s3_async.py -k move_null_version_onto_key -v
+```
+
+These nine cases belong to `TestS3FileSystem` and `TestAioS3FileSystem` and use
+their existing filesystem fixtures. The shared session-scoped bucket fixture in
+`tests/pyathena/filesystem/conftest.py` provisions the three buckets once per
+worker, so a serial run shares one setup and wait across both test modules.
+The synchronous API, asynchronous API, and asynchronous filesystem's synchronous
+wrapper each have a separate test method.
+
+The test identity needs `s3:CreateBucket`, `s3:GetBucketVersioning`,
+`s3:PutBucketVersioning`, `s3:PutObject`, `s3:GetObject`, `s3:GetObjectVersion`,
+`s3:ListBucketVersions`, `s3:DeleteObjectVersion`, and `s3:DeleteBucket` on the
+temporary `pyathena-mv-null-*` buckets and their objects, in addition to the normal
+test session permissions. The fixture deletes all versions and delete markers,
+then removes its buckets, including when a test fails. An interrupted process may
+leave buckets behind; remove only the buckets created by that run.
+
+Ordinary CI does not enable these tests. Its state-specific mock regression tests
+and existing AWS suite still run; report the opt-in results separately.
+
 ### SQLAlchemy suites
 
 The SQLAlchemy compliance suites under `tests/sqlalchemy/` run with different configurations: use `sqla` for synchronous dialects and `sqla-async` for native asyncio dialects.

pyathena/filesystem/s3.py

@@ -1414,9 +1414,17 @@ 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, 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.
+        destination is the file itself is left in place, and directories,
+        which S3 does not store as objects, are not copied. A ``null`` version
+        moved onto its key is copied and deleted only when bucket versioning
+        is enabled. Without versioning, or with versioning suspended, that
+        move is left in place because the copy would replace the version
+        that the deletion removes.
+
+        Comparing a ``null`` version with its unversioned key calls
+        GetBucketVersioning once per bucket during the move's planning and
+        requires ``s3:GetBucketVersioning``. The result is not cached across
+        moves. Directory buckets do not support versioning and need no lookup.
 
         Args:
             path1: Source S3 path, glob pattern, or list of paths.
@@ -1431,6 +1439,7 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
                 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.
+            OSError: If bucket versioning cannot be read.
         """
         if path1 == path2:
             return
@@ -1462,15 +1471,35 @@ def _move_pairs(
 
         Raises:
             ValueError: If the move has conflicting paths.
+            OSError: If bucket versioning cannot be read.
         """
         pairing = S3PathPairing(path1, path2, recursive=recursive, maxdepth=maxdepth)
         pairs = self._copy_pairs(pairing)
+        paths = {p: S3Path.parse(p) for pair in pairs for p in pair}
+        unversioned = {p.name for p in paths.values() if not p.version_id}
+        buckets = {
+            p.bucket
+            for p in paths.values()
+            if p.version_id == "null"
+            and p.name in unversioned
+            and not self.core._is_directory_bucket(p.bucket)
+        }
+        versioned_buckets = {
+            bucket
+            for bucket in buckets
+            if self._call(self._client.get_bucket_versioning, Bucket=bucket).get("Status")
+            == "Enabled"
+        }
         missing = {
             source
-            for source in pairing.conflict_candidates(pairs)
+            for source in pairing.conflict_candidates(
+                pairs, versioning_enabled_buckets=versioned_buckets
+            )
             if self._head_object(source) is None
         }
-        return pairing.move_pairs(pairs, missing=missing)
+        return pairing.move_pairs(
+            pairs, missing=missing, versioning_enabled_buckets=versioned_buckets
+        )
 
     def _copy_pairs(
         self, pairing: S3PathPairing, isdir: Callable[[str], bool] | None = None

pyathena/filesystem/s3_async.py

@@ -369,6 +369,7 @@ async def _mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> N
                 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.
+            OSError: If bucket versioning cannot be read.
         """
         if path1 == path2:
             return
@@ -400,6 +401,10 @@ def mv(self, path1, path2, recursive=False, maxdepth=None, **kwargs) -> None:
             recursive: Whether to move the directories with their contents.
             maxdepth: Maximum depth of a recursive move.
             **kwargs: Additional S3 copy parameters.
+
+        Raises:
+            ValueError: If the move has conflicting paths.
+            OSError: If bucket versioning cannot be read.
         """
         sync(self.loop, self._mv, path1, path2, recursive=recursive, maxdepth=maxdepth, **kwargs)
 

pyathena/filesystem/s3_path.py

@@ -135,13 +135,13 @@ def uri(self) -> str:
 
     @property
     def target(self) -> S3Path:
-        """What the path names when paths are compared: a ``null`` version names its key.
+        """The default move target, assuming bucket versioning is not enabled.
 
         A ``null`` version is taken to name the key itself: in a bucket without
         versioning, or with versioning suspended, a write to the key replaces
         its ``null`` version. With versioning enabled, a write adds a new
-        version instead, but the ``null`` version is still taken to name the
-        key. Any other path is its own target.
+        version instead, so callers with that state must keep the ``null``
+        version distinct from its key. Any other path is its own target.
         """
         return self.with_version_id(None) if self.version_id == "null" else self
 

pyathena/filesystem/s3_path_pairing.py

@@ -28,7 +28,9 @@ class S3PathPairing:
     pairs them, except that a path with a version ID names that version of an
     object: it is not a glob pattern, and its destination is named after its
     key without the version. A move compares the paths by what they name
-    (see :attr:`~pyathena.filesystem.s3_path.S3Path.target`).
+    (see :attr:`~pyathena.filesystem.s3_path.S3Path.target`), keeping ``null``
+    versions distinct from their keys in the caller's versioning-enabled
+    buckets.
     :meth:`delete_paths` splits the paths of an ``rm()``.
 
     The pairing is a frozen dataclass of the paths as given, and holds no
@@ -138,7 +140,12 @@ def copy_pairs(
         destinations = other_paths(names, path2, exists=exists, flatten=not source_is_str)
         return list(zip(sources, destinations, strict=True))
 
-    def conflict_candidates(self, pairs: Sequence[tuple[str, str]]) -> list[str]:
+    def conflict_candidates(
+        self,
+        pairs: Sequence[tuple[str, str]],
+        *,
+        versioning_enabled_buckets: Collection[str] = (),
+    ) -> list[str]:
         """Return the sources of a move whose conflicts depend on an object at their key.
 
         A source with another source below it may be a directory. If no object
@@ -149,18 +156,28 @@ def conflict_candidates(self, pairs: Sequence[tuple[str, str]]) -> list[str]:
         Args:
             pairs: The sources and destinations of the move, as
                 :meth:`copy_pairs` of this pairing returns them.
+            versioning_enabled_buckets: Buckets whose versioning is enabled.
+                Their ``null`` versions are distinct from their unversioned
+                keys. Use the same collection for :meth:`move_pairs`.
 
         Returns:
             The sources, in ``bucket/key`` form (their
             :attr:`~pyathena.filesystem.s3_path.S3Path.target`), that have
             another source below them and a destination that conflicts, one
             per pair in the order of the pairs; empty if nothing needs to be
             looked up.
+
+        Raises:
+            TypeError: If ``versioning_enabled_buckets`` is a string.
         """
-        return self._moves(pairs)[2]
+        return self._moves(pairs, versioning_enabled_buckets)[2]
 
     def move_pairs(
-        self, pairs: Sequence[tuple[str, str]], missing: Collection[str] | None = None
+        self,
+        pairs: Sequence[tuple[str, str]],
+        missing: Collection[str] | None = None,
+        *,
+        versioning_enabled_buckets: Collection[str] = (),
     ) -> list[tuple[str, str]]:
         """Check the pairs of a move and leave out the sources that stay in place.
 
@@ -170,26 +187,31 @@ def move_pairs(
             missing: The :meth:`conflict_candidates` without an object at their
                 key, in any form that names them; needed when there are
                 candidates.
+            versioning_enabled_buckets: Buckets whose versioning is enabled.
+                Their ``null`` versions are distinct from their unversioned
+                keys. Use the same collection for :meth:`conflict_candidates`.
 
         Returns:
             The pairs, except those whose destination is the source itself or,
-            for a ``null`` version, the key of the source.
+            for a ``null`` version in a bucket without enabled versioning,
+            the key of the source.
 
         Raises:
             ValueError: If two sources have the same destination, or a
                 destination is another source, including one left in place,
                 except for a directory with no object at its key, which is not
                 copied. Also if ``missing`` is needed and None.
             TypeError: If ``missing`` is a string instead of a collection.
+                Also if ``versioning_enabled_buckets`` is a string.
         """
         if isinstance(missing, str):
             raise TypeError("missing is a collection of paths, not a path.")
-        named, sources, candidates = self._moves(pairs)
+        named, sources, candidates = self._moves(pairs, versioning_enabled_buckets)
         if missing is None and candidates:
             raise ValueError("missing is needed to check the pairs.")
         # A directory without an object at its key writes no destination.
         skipped = set(candidates).intersection(
-            str(S3Path.parse(path).target) for path in missing or ()
+            self._target(S3Path.parse(path), versioning_enabled_buckets) for path in missing or ()
         )
         # A path with a version always names an object, so it writes its
         # destination even when the key has no current object.
@@ -244,21 +266,44 @@ def _is_glob(path: str) -> bool:
         """
         return has_magic(path) and not S3Path.has_version_id(path)
 
+    @staticmethod
+    def _target(path: S3Path, versioning_enabled_buckets: Collection[str]) -> str:
+        """Return a move target using the caller's bucket versioning state.
+
+        Args:
+            path: The path of the source or destination.
+            versioning_enabled_buckets: Buckets whose versioning is enabled.
+
+        Returns:
+            The normalized path, retaining a ``null`` version only when its
+            bucket has versioning enabled.
+        """
+        return str(path if path.bucket in versioning_enabled_buckets else path.target)
+
     @staticmethod
     def _moves(
         pairs: Sequence[tuple[str, str]],
+        versioning_enabled_buckets: Collection[str] = (),
     ) -> tuple[list[tuple[str, str, bool, str, str]], set[str], list[str]]:
         """Compare the paths of a move by what they name.
 
         Args:
             pairs: The sources and destinations of the move.
+            versioning_enabled_buckets: Buckets whose versioning is enabled.
 
         Returns:
             Each pair with whether its source has a version and the targets
             of its source and destination; the targets of all sources,
             including those left in place; and the conflict candidates (see
             :meth:`conflict_candidates`).
+
+        Raises:
+            TypeError: If ``versioning_enabled_buckets`` is a string.
         """
+        if isinstance(versioning_enabled_buckets, str):
+            raise TypeError(
+                "versioning_enabled_buckets is a collection of bucket names, not a bucket name."
+            )
         named = []
         for p1, p2 in pairs:
             source_path = S3Path.parse(p1)
@@ -267,8 +312,8 @@ def _moves(
                     p1,
                     p2,
                     bool(source_path.version_id),
-                    str(source_path.target),
-                    str(S3Path.parse(p2).target),
+                    S3PathPairing._target(source_path, versioning_enabled_buckets),
+                    S3PathPairing._target(S3Path.parse(p2), versioning_enabled_buckets),
                 )
             )
         # The sources left in place count too; a copy onto one overwrites it.