Add support for registering views (#3288)

# Rationale for this change

- Relates to apache/iceberg#14869

## Are these changes tested?

tests/catalog/test_rest.py contains new tests.

## Are there any user-facing changes?

This PR adds a new `register_view` method to `catalog`. 

<!-- In the case of user-facing changes, please add the changelog label.
-->

Author: ebyhr

mkdocs/docs/api.md

@@ -1529,6 +1529,17 @@ catalog = load_catalog("default")
 catalog.view_exists("default.bar")
 ```
 
+## Register a view
+
+To register a view using existing metadata:
+
+```python
+catalog.register_view(
+    identifier="docs_example.bids",
+    metadata_location="s3://warehouse/path/to/metadata.json"
+)
+```
+
 ## Table Statistics Management
 
 Manage table statistics with operations through the `Table` API:

pyiceberg/catalog/__init__.py

@@ -690,6 +690,22 @@ def update_namespace_properties(
             ValueError: If removals and updates have overlapping keys.
         """
 
+    @abstractmethod
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        """Register a new view using existing metadata.
+
+        Args:
+            identifier (Union[str, Identifier]): View identifier for the view
+            metadata_location (str): The location to the metadata
+
+        Returns:
+            View: The newly registered view
+
+        Raises:
+            ViewAlreadyExistsError: If the view already exists.
+            TableAlreadyExistsError: If a table with the same name already exists.
+        """
+
     @abstractmethod
     def drop_view(self, identifier: str | Identifier) -> None:
         """Drop a view.

pyiceberg/catalog/bigquery_metastore.py

@@ -305,6 +305,9 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o
     def list_views(self, namespace: str | Identifier) -> list[Identifier]:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/dynamodb.py

@@ -553,6 +553,9 @@ def create_view(
     def list_views(self, namespace: str | Identifier) -> list[Identifier]:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/glue.py

@@ -970,6 +970,9 @@ def create_view(
     def list_views(self, namespace: str | Identifier) -> list[Identifier]:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/hive.py

@@ -854,6 +854,9 @@ def update_namespace_properties(
 
         return PropertiesUpdateSummary(removed=list(removed or []), updated=list(updated or []), missing=list(expected_to_change))
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/noop.py

@@ -132,6 +132,9 @@ def view_exists(self, identifier: str | Identifier) -> bool:
     def namespace_exists(self, namespace: str | Identifier) -> bool:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

pyiceberg/catalog/rest/__init__.py

@@ -154,6 +154,7 @@ class Endpoints:
     list_views: str = "namespaces/{namespace}/views"
     load_view: str = "namespaces/{namespace}/views/{view}"
     create_view: str = "namespaces/{namespace}/views"
+    register_view: str = "namespaces/{namespace}/register-view"
     drop_view: str = "namespaces/{namespace}/views/{view}"
     view_exists: str = "namespaces/{namespace}/views/{view}"
     plan_table_scan: str = "namespaces/{namespace}/tables/{table}/plan"
@@ -183,6 +184,7 @@ class Capability:
     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_REGISTER_VIEW = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.register_view}")
     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}")
     V1_TABLE_SCAN_PLAN_TASKS = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.fetch_scan_tasks}")
@@ -322,6 +324,11 @@ class RegisterTableRequest(IcebergBaseModel):
     overwrite: bool
 
 
+class RegisterViewRequest(IcebergBaseModel):
+    name: str
+    metadata_location: str = Field(..., alias="metadata-location")
+
+
 class ConfigResponse(IcebergBaseModel):
     defaults: Properties | None = Field(default_factory=dict)
     overrides: Properties | None = Field(default_factory=dict)
@@ -1332,6 +1339,29 @@ def view_exists(self, identifier: str | Identifier) -> bool:
 
         return False
 
+    @retry(**_RETRY_ARGS)
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        self._check_endpoint(Capability.V1_REGISTER_VIEW)
+        namespace_and_view = self._split_identifier_for_path(identifier, IdentifierKind.VIEW)
+        namespace = namespace_and_view["namespace"]
+        view = namespace_and_view["view"]
+        if self.table_exists(identifier):
+            raise TableAlreadyExistsError(f"Table {namespace}.{view} already exists")
+
+        request = RegisterViewRequest(name=view, metadata_location=metadata_location)
+        serialized_json = request.model_dump_json().encode(UTF8)
+        response = self._session.post(
+            self.url(Endpoints.register_view, namespace=namespace),
+            data=serialized_json,
+        )
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            _handle_non_200_response(exc, {409: ViewAlreadyExistsError})
+
+        view_response = ViewResponse.model_validate_json(response.text)
+        return self._response_to_view(self.identifier_to_tuple(identifier), view_response)
+
     @retry(**_RETRY_ARGS)
     def drop_view(self, identifier: str) -> None:
         self._check_endpoint(Capability.V1_DELETE_VIEW)

pyiceberg/catalog/sql.py

@@ -745,6 +745,9 @@ def list_views(self, namespace: str | Identifier) -> list[Identifier]:
     def view_exists(self, identifier: str | Identifier) -> bool:
         raise NotImplementedError
 
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 

tests/catalog/test_rest.py

@@ -105,6 +105,7 @@
     Capability.V1_LIST_VIEWS,
     Capability.V1_LOAD_VIEW,
     Capability.V1_VIEW_EXISTS,
+    Capability.V1_REGISTER_VIEW,
     Capability.V1_DELETE_VIEW,
     Capability.V1_SUBMIT_TABLE_SCAN_PLAN,
     Capability.V1_TABLE_SCAN_PLAN_TASKS,
@@ -2182,6 +2183,72 @@ def test_table_identifier_in_commit_table_request(
     )
 
 
+def test_register_view_200(rest_mock: Mocker, example_view_metadata_rest_json: dict[str, Any]) -> None:
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/default/tables/registered_view",
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces/default/register-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.register_view(
+        identifier=("default", "registered_view"), metadata_location="s3://warehouse/database/view/metadata.json"
+    )
+    expected = View(
+        identifier=("default", "registered_view"),
+        metadata=ViewMetadata(**example_view_metadata_rest_json["metadata"]),
+    )
+    assert actual == expected
+
+
+def test_register_view_409_view(rest_mock: Mocker) -> None:
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/default/tables/registered_view",
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces/default/register-view",
+        json={
+            "error": {
+                "message": "View already exists: default.view in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+                "type": "AlreadyExistsException",
+                "code": 409,
+            }
+        },
+        status_code=409,
+        request_headers=TEST_HEADERS,
+    )
+
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    with pytest.raises(ViewAlreadyExistsError) as e:
+        catalog.register_view(
+            identifier=("default", "registered_view"), metadata_location="s3://warehouse/database/view/metadata.json"
+        )
+    assert "View already exists" in str(e.value)
+
+
+def test_register_view_409_table(rest_mock: Mocker) -> None:
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/default/tables/registered_view",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    with pytest.raises(TableAlreadyExistsError) as e:
+        catalog.register_view(
+            identifier=("default", "registered_view"), metadata_location="s3://warehouse/database/view/metadata.json"
+        )
+    assert "Table default.registered_view already exists" in str(e.value)
+
+
 def test_drop_view_invalid_namespace(rest_mock: Mocker) -> None:
     view = "view"
     with pytest.raises(NoSuchIdentifierError) as e: