Fix exists() exception narrowing and add real HF revision integration test

FsspecInputFile.exists() reimplemented exists()-via-info() with a
try/except catching only FileNotFoundError, unlike fsspec's own
lexists()/exists() which swallow any exception. Delegate to
self._fs.exists() directly instead, which already forwards kwargs to
info() with the correct broad exception handling.

Also add an HF_TOKEN-gated integration test against a real (temporary)
Hugging Face dataset repo, demonstrating the concrete problem this
property solves: without hf.revision, reads always follow the repo's
moving default branch, so a file "written" at one commit silently
returns different content once someone pushes a new commit to the
same path. Minor: use a _HF_SCHEMES frozenset in _get_fs_kwargs for
consistency with the existing _ADLS_SCHEMES dispatch pattern, and
de-duplicate/strengthen two tests that asserted only on the private
_fs_kwargs attribute.

Author: kszucs

pyiceberg/io/fsspec.py

@@ -320,6 +320,7 @@ def _hf(properties: Properties) -> AbstractFileSystem:
 }
 
 _ADLS_SCHEMES = frozenset({"abfs", "abfss", "wasb", "wasbs"})
+_HF_SCHEMES = frozenset({"hf"})
 
 
 class FsspecInputFile(InputFile):
@@ -351,13 +352,11 @@ def __len__(self) -> int:
     def exists(self) -> bool:
         """Check whether the location exists."""
         if self._fs_kwargs:
-            # fsspec's AbstractFileSystem.lexists() doesn't forward **kwargs to exists()/info(),
-            # so honoring fs_kwargs (e.g. a pinned HuggingFace Hub revision) requires calling info() directly.
-            try:
-                self._fs.info(self.location, **self._fs_kwargs)
-                return True
-            except FileNotFoundError:
-                return False
+            # fsspec's AbstractFileSystem.lexists() doesn't forward **kwargs to exists()/info(), so
+            # honoring fs_kwargs (e.g. a pinned HuggingFace Hub revision) requires calling exists()
+            # directly -- it does forward kwargs to info(), with the same broad exception handling
+            # lexists() would otherwise provide.
+            return self._fs.exists(self.location, **self._fs_kwargs)
         return self._fs.lexists(self.location)
 
     @override
@@ -503,7 +502,7 @@ def _get_fs_kwargs(self, scheme: str) -> Properties:
         commit, neither of which is a valid write target, so writes/deletes always target the
         repository's default (writable) branch.
         """
-        if scheme == "hf" and (revision := self.properties.get(HF_REVISION)):
+        if scheme in _HF_SCHEMES and (revision := self.properties.get(HF_REVISION)):
             return {"revision": revision}
         return {}
 

tests/io/test_fsspec.py

@@ -1134,6 +1134,7 @@ def test_fsspec_hf_revision_forwarded_to_reads() -> None:
     with mock.patch("huggingface_hub.HfFileSystem") as mock_hf_fs:
         mock_fs = mock_hf_fs.return_value
         mock_fs.info.return_value = {"size": 123}
+        mock_fs.exists.return_value = True
 
         hf_fileio = FsspecFileIO(properties=session_properties)
         input_file = hf_fileio.new_input(location=location)
@@ -1142,7 +1143,7 @@ def test_fsspec_hf_revision_forwarded_to_reads() -> None:
         mock_fs.info.assert_called_with(location, revision="a-pinned-revision")
 
         assert input_file.exists() is True
-        mock_fs.info.assert_called_with(location, revision="a-pinned-revision")
+        mock_fs.exists.assert_called_with(location, revision="a-pinned-revision")
 
         input_file.open()
         mock_fs.open.assert_called_with(location, "rb", revision="a-pinned-revision")
@@ -1166,11 +1167,17 @@ def test_fsspec_hf_revision_not_forwarded_to_writes() -> None:
 
 
 def test_fsspec_hf_no_revision_by_default() -> None:
-    with mock.patch("huggingface_hub.HfFileSystem"):
+    location = "hf://datasets/user/repo/file.parquet"
+
+    with mock.patch("huggingface_hub.HfFileSystem") as mock_hf_fs:
+        mock_fs = mock_hf_fs.return_value
+        mock_fs.info.return_value = {"size": 123}
+
         hf_fileio = FsspecFileIO(properties={})
-        input_file = hf_fileio.new_input(location="hf://datasets/user/repo/file.parquet")
+        input_file = hf_fileio.new_input(location=location)
 
-        assert input_file._fs_kwargs == {}
+        assert len(input_file) == 123
+        mock_fs.info.assert_called_with(location)
 
 
 def test_fsspec_non_hf_scheme_does_not_receive_revision_kwarg() -> None:
@@ -1182,3 +1189,38 @@ def test_fsspec_non_hf_scheme_does_not_receive_revision_kwarg() -> None:
     input_file = fileio.new_input(location="file:///tmp/foo.parquet")
 
     assert input_file._fs_kwargs == {}
+
+
+@pytest.mark.skipif(not os.environ.get("HF_TOKEN"), reason="Requires a real Hugging Face Hub token in HF_TOKEN")
+def test_fsspec_hf_revision_pins_reads_to_a_fixed_commit() -> None:
+    """Without hf.revision, reads always follow the repo's current default branch.
+
+    This means an Iceberg data-file location pointing into an `hf://` repo isn't reproducible on
+    its own: if someone pushes a new commit that changes the same path, every future read of that
+    "immutable" data file silently returns the new content instead of what existed when the
+    Iceberg snapshot referencing it was written. hf.revision fixes this by letting a table pin
+    reads to the exact commit that was current when the file was written.
+    """
+    from huggingface_hub import HfApi
+
+    hf_token = os.environ["HF_TOKEN"]
+    api = HfApi(token=hf_token)
+    repo_id = f"pyiceberg-hf-revision-test-{uuid.uuid4().hex[:8]}"
+    api.create_repo(repo_id=repo_id, repo_type="dataset")  # private by default
+    location = f"hf://datasets/{repo_id}/file.txt"
+
+    try:
+        first_commit = api.upload_file(path_or_fileobj=b"v1", path_in_repo="file.txt", repo_id=repo_id, repo_type="dataset")
+        api.upload_file(path_or_fileobj=b"v2", path_in_repo="file.txt", repo_id=repo_id, repo_type="dataset")
+
+        # Without hf.revision, reads follow the moving default branch -- the file now reads "v2",
+        # even though an Iceberg data file referencing it was "written" back when it was "v1".
+        default_branch_fileio = FsspecFileIO(properties={"hf.token": hf_token})
+        assert default_branch_fileio.new_input(location).open().read() == b"v2"
+
+        # Pinning hf.revision to the first commit makes the read reproducible: it keeps returning
+        # "v1" regardless of what's since been pushed to the default branch.
+        pinned_fileio = FsspecFileIO(properties={"hf.token": hf_token, "hf.revision": first_commit.oid})
+        assert pinned_fileio.new_input(location).open().read() == b"v1"
+    finally:
+        api.delete_repo(repo_id=repo_id, repo_type="dataset")