From 412197dc6a4cee3a33b57753f24808a753227abc Mon Sep 17 00:00:00 2001
From: spr0els <58633269+spr0els@users.noreply.github.com>
Date: Thu, 9 Apr 2026 15:02:48 +0200
Subject: [PATCH 1/5] feat: add load_view to REST catalog

---
 pyiceberg/catalog/__init__.py           | 17 +++++++++++++
 pyiceberg/catalog/bigquery_metastore.py |  4 +++
 pyiceberg/catalog/dynamodb.py           |  3 +++
 pyiceberg/catalog/glue.py               |  3 +++
 pyiceberg/catalog/hive.py               |  3 +++
 pyiceberg/catalog/noop.py               |  3 +++
 pyiceberg/catalog/rest/__init__.py      | 18 ++++++++++++++
 pyiceberg/catalog/sql.py                |  3 +++
 tests/catalog/test_rest.py              | 33 +++++++++++++++++++++++++
 9 files changed, 87 insertions(+)

diff --git a/pyiceberg/catalog/__init__.py b/pyiceberg/catalog/__init__.py
index beb02b8059..fc6f80f458 100644
--- a/pyiceberg/catalog/__init__.py
+++ b/pyiceberg/catalog/__init__.py
@@ -646,6 +646,23 @@ def list_views(self, namespace: str | Identifier) -> list[Identifier]:
             NoSuchNamespaceError: If a namespace with the given name does not exist.
         """
 
+    @abstractmethod
+    def load_view(self, identifier: str | Identifier) -> View:
+        """Load the view's metadata and returns the view instance.
+
+        You can also use this method to check for view existence using 'try catalog.load_view() except NoSuchViewError'.
+        Note: This method doesn't scan data stored in the view.
+
+        Args:
+            identifier (str | Identifier): View identifier.
+
+        Returns:
+            View: the view instance with its metadata.
+
+        Raises:
+            NoSuchViewError: If a view with the name does not exist.
+        """
+
     @abstractmethod
     def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
         """Get properties for a namespace.
diff --git a/pyiceberg/catalog/bigquery_metastore.py b/pyiceberg/catalog/bigquery_metastore.py
index 6568e29aed..3fe99f3f02 100644
--- a/pyiceberg/catalog/bigquery_metastore.py
+++ b/pyiceberg/catalog/bigquery_metastore.py
@@ -41,6 +41,7 @@
 from pyiceberg.table.update import TableRequirement, TableUpdate
 from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties
 from pyiceberg.utils.config import Config
+from pyiceberg.view import View
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -314,6 +315,9 @@ def drop_view(self, identifier: str | Identifier) -> None:
     def view_exists(self, identifier: str | Identifier) -> bool:
         raise NotImplementedError
 
+    def load_view(self, identifier: str | Identifier) -> View:
+        raise NotImplementedError
+
     def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
         dataset_name = self.identifier_to_database(namespace)
 
diff --git a/pyiceberg/catalog/dynamodb.py b/pyiceberg/catalog/dynamodb.py
index a4d900e160..aa66f08bcf 100644
--- a/pyiceberg/catalog/dynamodb.py
+++ b/pyiceberg/catalog/dynamodb.py
@@ -559,6 +559,9 @@ def drop_view(self, identifier: str | Identifier) -> None:
     def view_exists(self, identifier: str | Identifier) -> bool:
         raise NotImplementedError
 
+    def load_view(self, identifier: str | Identifier) -> View:
+        raise NotImplementedError
+
     def _get_iceberg_table_item(self, database_name: str, table_name: str) -> dict[str, Any]:
         try:
             return self._get_dynamo_item(identifier=f"{database_name}.{table_name}", namespace=database_name)
diff --git a/pyiceberg/catalog/glue.py b/pyiceberg/catalog/glue.py
index 81c4c57d6a..a21fe6da3c 100644
--- a/pyiceberg/catalog/glue.py
+++ b/pyiceberg/catalog/glue.py
@@ -976,6 +976,9 @@ def drop_view(self, identifier: str | Identifier) -> None:
     def view_exists(self, identifier: str | Identifier) -> bool:
         raise NotImplementedError
 
+    def load_view(self, identifier: str | Identifier) -> View:
+        raise NotImplementedError
+
     @staticmethod
     def __is_iceberg_table(table: "TableTypeDef") -> bool:
         return table.get("Parameters", {}).get(TABLE_TYPE, "").lower() == ICEBERG
diff --git a/pyiceberg/catalog/hive.py b/pyiceberg/catalog/hive.py
index 3453f9266a..afca5954b1 100644
--- a/pyiceberg/catalog/hive.py
+++ b/pyiceberg/catalog/hive.py
@@ -486,6 +486,9 @@ def list_views(self, namespace: str | Identifier) -> list[Identifier]:
     def view_exists(self, identifier: str | Identifier) -> bool:
         raise NotImplementedError
 
+    def load_view(self, identifier: str | Identifier) -> View:
+        raise NotImplementedError
+
     def _create_lock_request(self, database_name: str, table_name: str) -> LockRequest:
         lock_component: LockComponent = LockComponent(
             level=LockLevel.TABLE, type=LockType.EXCLUSIVE, dbname=database_name, tablename=table_name, isTransactional=True
diff --git a/pyiceberg/catalog/noop.py b/pyiceberg/catalog/noop.py
index cb714e1c5a..1524343610 100644
--- a/pyiceberg/catalog/noop.py
+++ b/pyiceberg/catalog/noop.py
@@ -144,3 +144,6 @@ def create_view(
         properties: Properties = EMPTY_DICT,
     ) -> View:
         raise NotImplementedError
+
+    def load_view(self, identifier: str | Identifier) -> View:
+        raise NotImplementedError
diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py
index ca0ff75e8c..299e95a41d 100644
--- a/pyiceberg/catalog/rest/__init__.py
+++ b/pyiceberg/catalog/rest/__init__.py
@@ -152,6 +152,7 @@ class Endpoints:
     get_token: str = "oauth/tokens"
     rename_table: str = "tables/rename"
     list_views: str = "namespaces/{namespace}/views"
+    load_view: str = "namespaces/{namespace}/views/{view}"
     create_view: str = "namespaces/{namespace}/views"
     drop_view: str = "namespaces/{namespace}/views/{view}"
     view_exists: str = "namespaces/{namespace}/views/{view}"
@@ -180,6 +181,7 @@ class Capability:
     V1_REGISTER_TABLE = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.register_table}")
 
     V1_LIST_VIEWS = Endpoint(http_method=HttpMethod.GET, path=f"{API_PREFIX}/{Endpoints.list_views}")
+    V1_LOAD_VIEW = Endpoint(http_method=HttpMethod.GET, path=f"{API_PREFIX}/{Endpoints.load_view}")
     V1_VIEW_EXISTS = Endpoint(http_method=HttpMethod.HEAD, path=f"{API_PREFIX}/{Endpoints.view_exists}")
     V1_DELETE_VIEW = Endpoint(http_method=HttpMethod.DELETE, path=f"{API_PREFIX}/{Endpoints.drop_view}")
     V1_SUBMIT_TABLE_SCAN_PLAN = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.plan_table_scan}")
@@ -209,6 +211,7 @@ class Capability:
 VIEW_ENDPOINTS: frozenset[Endpoint] = frozenset(
     (
         Capability.V1_LIST_VIEWS,
+        Capability.V1_LOAD_VIEW,
         Capability.V1_DELETE_VIEW,
     )
 )
@@ -1109,6 +1112,21 @@ def list_views(self, namespace: str | Identifier) -> list[Identifier]:
             _handle_non_200_response(exc, {404: NoSuchNamespaceError})
         return [(*view.namespace, view.name) for view in ListViewsResponse.model_validate_json(response.text).identifiers]
 
+    @retry(**_RETRY_ARGS)
+    def load_view(self, identifier: str | Identifier) -> View:
+        self._check_endpoint(Capability.V1_LOAD_VIEW)
+        response = self._session.get(
+            self.url(Endpoints.load_view, prefixed=True, **self._split_identifier_for_path(identifier, IdentifierKind.VIEW)),
+            params={},
+        )
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            _handle_non_200_response(exc, {404: NoSuchViewError})
+
+        view_response = ViewResponse.model_validate_json(response.text)
+        return self._response_to_view(self.identifier_to_tuple(identifier), view_response)
+
     @retry(**_RETRY_ARGS)
     def commit_table(
         self, table: Table, requirements: tuple[TableRequirement, ...], updates: tuple[TableUpdate, ...]
diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py
index a65b5f0d5d..dc703e95a4 100644
--- a/pyiceberg/catalog/sql.py
+++ b/pyiceberg/catalog/sql.py
@@ -748,6 +748,9 @@ def view_exists(self, identifier: str | Identifier) -> bool:
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 
+    def load_view(self, identifier: str | Identifier) -> View:
+        raise NotImplementedError
+
     def close(self) -> None:
         """Close the catalog and release database connections.
 
diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py
index 7977892635..7adab0c8bb 100644
--- a/tests/catalog/test_rest.py
+++ b/tests/catalog/test_rest.py
@@ -103,6 +103,7 @@
     Capability.V1_RENAME_TABLE,
     Capability.V1_REGISTER_TABLE,
     Capability.V1_LIST_VIEWS,
+    Capability.V1_LOAD_VIEW,
     Capability.V1_VIEW_EXISTS,
     Capability.V1_DELETE_VIEW,
     Capability.V1_SUBMIT_TABLE_SCAN_PLAN,
@@ -1449,6 +1450,38 @@ def test_create_view_409(
     assert "View already exists" in str(e.value)
 
 
+def test_load_view_200(rest_mock: Mocker, example_view_metadata_rest_json: dict[str, Any]) -> None:
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/fokko/views/view",
+        json=example_view_metadata_rest_json,
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    actual = catalog.load_view(("fokko", "view"))
+    expected = View(identifier=("fokko", "view"), metadata=ViewMetadata(**example_view_metadata_rest_json["metadata"]))
+    assert actual == expected
+
+
+def test_load_view_404(rest_mock: Mocker) -> None:
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/fokko/views/non_existent_view",
+        json={
+            "error": {
+                "message": "View does not exist: examples.non_existent_view in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+                "type": "NoSuchNamespaceErrorException",
+                "code": 404,
+            }
+        },
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+
+    with pytest.raises(NoSuchViewError) as e:
+        RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).load_view(("fokko", "non_existent_view"))
+    assert "View does not exist" in str(e.value)
+
+
 def test_create_table_if_not_exists_200(
     rest_mock: Mocker, table_schema_simple: Schema, example_table_metadata_no_snapshot_v1_rest_json: dict[str, Any]
 ) -> None:

From af07ae0a98dea051a2397c4d91694a7beffc7a70 Mon Sep 17 00:00:00 2001
From: spr0els <58633269+spr0els@users.noreply.github.com>
Date: Wed, 15 Apr 2026 15:17:19 +0200
Subject: [PATCH 2/5] refactor: cleanup docstring and method call parameters in
 load_view

---
 pyiceberg/catalog/__init__.py      | 3 ---
 pyiceberg/catalog/rest/__init__.py | 3 +--
 2 files changed, 1 insertion(+), 5 deletions(-)

diff --git a/pyiceberg/catalog/__init__.py b/pyiceberg/catalog/__init__.py
index fc6f80f458..5db35ac3bd 100644
--- a/pyiceberg/catalog/__init__.py
+++ b/pyiceberg/catalog/__init__.py
@@ -650,9 +650,6 @@ def list_views(self, namespace: str | Identifier) -> list[Identifier]:
     def load_view(self, identifier: str | Identifier) -> View:
         """Load the view's metadata and returns the view instance.
 
-        You can also use this method to check for view existence using 'try catalog.load_view() except NoSuchViewError'.
-        Note: This method doesn't scan data stored in the view.
-
         Args:
             identifier (str | Identifier): View identifier.
 
diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py
index 299e95a41d..b3a80e11aa 100644
--- a/pyiceberg/catalog/rest/__init__.py
+++ b/pyiceberg/catalog/rest/__init__.py
@@ -1116,8 +1116,7 @@ def list_views(self, namespace: str | Identifier) -> list[Identifier]:
     def load_view(self, identifier: str | Identifier) -> View:
         self._check_endpoint(Capability.V1_LOAD_VIEW)
         response = self._session.get(
-            self.url(Endpoints.load_view, prefixed=True, **self._split_identifier_for_path(identifier, IdentifierKind.VIEW)),
-            params={},
+            self.url(Endpoints.load_view, prefixed=True, **self._split_identifier_for_path(identifier, IdentifierKind.VIEW))
         )
         try:
             response.raise_for_status()

From 6af825aebdde64f50270f00190a099001c3b3943 Mon Sep 17 00:00:00 2001
From: spr0els <58633269+spr0els@users.noreply.github.com>
Date: Thu, 23 Apr 2026 17:08:33 +0200
Subject: [PATCH 3/5] test: add integration tests for load_view

---
 tests/conftest.py                      |  7 ++++
 tests/integration/test_rest_catalog.py | 49 ++++++++++++++++++++++++++
 2 files changed, 56 insertions(+)

diff --git a/tests/conftest.py b/tests/conftest.py
index d1a9f92886..b74e2ecabc 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -2353,6 +2353,13 @@ def table_list(table_name: str) -> list[str]:
     return [f"{table_name}_{idx}" for idx in range(NUM_TABLES)]
 
 
+@pytest.fixture()
+def view_name() -> str:
+    prefix = "my_iceberg_view-"
+    random_tag = "".join(choice(string.ascii_letters) for _ in range(RANDOM_LENGTH))
+    return (prefix + random_tag).lower()
+
+
 @pytest.fixture()
 def database_name() -> str:
     prefix = "my_iceberg_database-"
diff --git a/tests/integration/test_rest_catalog.py b/tests/integration/test_rest_catalog.py
index 18aa943175..f549e67420 100644
--- a/tests/integration/test_rest_catalog.py
+++ b/tests/integration/test_rest_catalog.py
@@ -16,10 +16,15 @@
 # under the License.
 # pylint:disable=redefined-outer-name
 
+import time
+
 import pytest
 from pytest_lazy_fixtures import lf
 
 from pyiceberg.catalog.rest import RestCatalog
+from pyiceberg.exceptions import NoSuchViewError
+from pyiceberg.schema import Schema
+from pyiceberg.view.metadata import SQLViewRepresentation, ViewVersion
 
 TEST_NAMESPACE_IDENTIFIER = "TEST NS"
 
@@ -62,3 +67,47 @@ def test_create_namespace_if_already_existing(catalog: RestCatalog) -> None:
     catalog.create_namespace_if_not_exists(TEST_NAMESPACE_IDENTIFIER)
 
     assert catalog.namespace_exists(TEST_NAMESPACE_IDENTIFIER)
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [lf("session_catalog")])
+def test_load_view(catalog: RestCatalog, table_schema_nested: Schema, database_name: str, view_name: str) -> None:
+    identifier = (database_name, view_name)
+    if not catalog.namespace_exists(database_name):
+        catalog.create_namespace(database_name)
+
+    view_version = ViewVersion(
+        version_id=1,
+        schema_id=1,
+        timestamp_ms=int(time.time() * 1000),
+        summary={},
+        representations=[
+            SQLViewRepresentation(
+                type="sql",
+                sql="SELECT 1 as some_col",
+                dialect="spark",
+            )
+        ],
+        default_namespace=["default"],
+    )
+    view = catalog.create_view(identifier, table_schema_nested, view_version=view_version)
+    loaded_view = catalog.load_view(identifier)
+    assert view.name() == loaded_view.name()
+    assert view.metadata == loaded_view.metadata
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [lf("session_catalog")])
+def test_load_view_with_table_ident(
+    catalog: RestCatalog, table_name: str, table_schema_nested: Schema, database_name: str
+) -> None:
+    table_identifier = (database_name, table_name)
+    if not catalog.namespace_exists(database_name):
+        catalog.create_namespace(database_name)
+
+    if not catalog.table_exists(table_identifier):
+        catalog.create_table(table_identifier, table_schema_nested)
+
+    assert catalog.table_exists(table_identifier)
+    with pytest.raises(NoSuchViewError):
+        catalog.load_view(table_identifier)

From 58526a355ab8811ef2927ae2fa729a227c7d184d Mon Sep 17 00:00:00 2001
From: spr0els <58633269+spr0els@users.noreply.github.com>
Date: Mon, 4 May 2026 14:47:57 +0200
Subject: [PATCH 4/5] fix: use correct type of exception in test_load_view_404

---
 tests/catalog/test_rest.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py
index 7adab0c8bb..271c942aa1 100644
--- a/tests/catalog/test_rest.py
+++ b/tests/catalog/test_rest.py
@@ -1469,7 +1469,7 @@ def test_load_view_404(rest_mock: Mocker) -> None:
         json={
             "error": {
                 "message": "View does not exist: examples.non_existent_view in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
-                "type": "NoSuchNamespaceErrorException",
+                "type": "NoSuchViewException",
                 "code": 404,
             }
         },

From 2047e16cfe82ec31073be02c3974acbc5f3b809b Mon Sep 17 00:00:00 2001
From: spr0els <58633269+spr0els@users.noreply.github.com>
Date: Tue, 5 May 2026 13:27:04 +0200
Subject: [PATCH 5/5] test: verify entire view object

Co-authored-by: Yuya Ebihara <ebyhry@gmail.com>
---
 tests/integration/test_rest_catalog.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/tests/integration/test_rest_catalog.py b/tests/integration/test_rest_catalog.py
index f549e67420..05039a982e 100644
--- a/tests/integration/test_rest_catalog.py
+++ b/tests/integration/test_rest_catalog.py
@@ -92,8 +92,7 @@ def test_load_view(catalog: RestCatalog, table_schema_nested: Schema, database_n
     )
     view = catalog.create_view(identifier, table_schema_nested, view_version=view_version)
     loaded_view = catalog.load_view(identifier)
-    assert view.name() == loaded_view.name()
-    assert view.metadata == loaded_view.metadata
+    assert view == loaded_view
 
 
 @pytest.mark.integration