Add snapshot-loading-mode to RESTCatalog

This will allow to set the snapshots to be send back. In case
of refs, only the snapshots referenced by branches or tags will
be returned:

https://github.com/apache/iceberg/blob/5d2230ead79da64a8c871a02eb1304a94aaece5c/open-api/rest-catalog-open-api.yaml#L954-L956

Author: Fokko

pyiceberg/catalog/rest/__init__.py

@@ -138,6 +138,7 @@ class IdentifierKind(Enum):
 SIGV4_REGION = "rest.signing-region"
 SIGV4_SERVICE = "rest.signing-name"
 OAUTH2_SERVER_URI = "oauth2-server-uri"
+SNAPSHOT_LOADING_MODE = "snapshot-loading-mode"
 
 NAMESPACE_SEPARATOR = b"\x1f".decode(UTF8)
 
@@ -678,7 +679,16 @@ def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
 
     @retry(**_RETRY_ARGS)
     def load_table(self, identifier: Union[str, Identifier]) -> Table:
-        response = self._session.get(self.url(Endpoints.load_table, prefixed=True, **self._split_identifier_for_path(identifier)))
+        params = {}
+        if mode := self.properties.get(SNAPSHOT_LOADING_MODE):
+            if mode in {"all", "refs"}:
+                params["snapshots"] = mode
+            else:
+                raise ValueError("Invalid snapshot-loading-mode: {}")
+
+        response = self._session.get(
+            self.url(Endpoints.load_table, prefixed=True, **self._split_identifier_for_path(identifier)), params=params
+        )
         try:
             response.raise_for_status()
         except HTTPError as exc:
@@ -816,7 +826,7 @@ def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identi
         try:
             response.raise_for_status()
         except HTTPError as exc:
-            self._handle_non_200_response(exc, {})
+            self._handle_non_200_response(exc, {404: NoSuchNamespaceError})
 
         return ListNamespaceResponse.model_validate_json(response.text).namespaces
 

tests/catalog/test_rest.py

@@ -24,7 +24,7 @@
 
 import pyiceberg
 from pyiceberg.catalog import PropertiesUpdateSummary, load_catalog
-from pyiceberg.catalog.rest import OAUTH2_SERVER_URI, RestCatalog
+from pyiceberg.catalog.rest import OAUTH2_SERVER_URI, SNAPSHOT_LOADING_MODE, RestCatalog
 from pyiceberg.exceptions import (
     AuthorizationExpiredError,
     NamespaceAlreadyExistsError,
@@ -555,6 +555,24 @@ def test_list_namespace_with_parent_200(rest_mock: Mocker) -> None:
     ]
 
 
+def test_list_namespace_with_parent_404(rest_mock: Mocker) -> None:
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces?parent=some_namespace",
+        json={
+            "error": {
+                "message": "Namespace provided in the `parent` query parameter is not found",
+                "type": "NoSuchNamespaceException",
+                "code": 404,
+            }
+        },
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+
+    with pytest.raises(NoSuchNamespaceError):
+        RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).list_namespaces(("some_namespace",))
+
+
 @pytest.mark.filterwarnings(
     "ignore:Deprecated in 0.8.0, will be removed in 1.0.0. Iceberg REST client is missing the OAuth2 server URI:DeprecationWarning"
 )
@@ -835,6 +853,29 @@ def test_load_table_200(rest_mock: Mocker, example_table_metadata_with_snapshot_
     assert actual == expected
 
 
+def test_load_table_200_loading_mode(
+    rest_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: Dict[str, Any]
+) -> None:
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/fokko/tables/table?snapshots=refs",
+        json=example_table_metadata_with_snapshot_v1_rest_json,
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN, **{SNAPSHOT_LOADING_MODE: "refs"})
+    actual = catalog.load_table(("fokko", "table"))
+    expected = Table(
+        identifier=("fokko", "table"),
+        metadata_location=example_table_metadata_with_snapshot_v1_rest_json["metadata-location"],
+        metadata=TableMetadataV1(**example_table_metadata_with_snapshot_v1_rest_json["metadata"]),
+        io=load_file_io(),
+        catalog=catalog,
+    )
+    # First compare the dicts
+    assert actual.metadata.model_dump() == expected.metadata.model_dump()
+    assert actual == expected
+
+
 def test_load_table_honor_access_delegation(
     rest_mock: Mocker, example_table_metadata_with_snapshot_v1_rest_json: Dict[str, Any]
 ) -> None: