Add rename_view to REST Catalog

mkdocs/docs/api.md

@@ -1746,6 +1746,17 @@ To list views in the `default` namespace:
 catalog.list_views("default")
 ```
 
+## Rename a view
+
+To rename a view:
+
+```python
+catalog.rename_view(
+    from_identifier="default.some_view",
+    to_identifier="default.some_view_renamed"
+)
+```
+
 ## Drop a view
 
 To drop a view:

pyiceberg/catalog/__init__.py

@@ -744,6 +744,20 @@ def create_view(
             ViewAlreadyExistsError: If a view with the name already exists.
         """
 
+    @abstractmethod
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+        """Rename a fully qualified view name.
+
+        Args:
+            from_identifier (str | Identifier): Existing view identifier.
+            to_identifier (str | Identifier): New view identifier.
+
+        Raises:
+            NoSuchViewError: If a view with the name does not exist.
+            ViewAlreadyExistsError: If a table or view with the target name already exists.
+            NoSuchNamespaceError: If the source or destination namespace does not exist.
+        """
+
     @staticmethod
     def identifier_to_tuple(identifier: str | Identifier) -> Identifier:
         """Parse an identifier to a tuple.

pyiceberg/catalog/bigquery_metastore.py

@@ -333,6 +333,10 @@ def view_exists(self, identifier: str | Identifier) -> bool:
     def load_view(self, identifier: str | Identifier) -> View:
         raise NotImplementedError
 
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+        raise NotImplementedError
+
     @override
     def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
         dataset_name = self.identifier_to_database(namespace, NoSuchNamespaceError)

pyiceberg/catalog/dynamodb.py

@@ -584,6 +584,10 @@ def view_exists(self, identifier: str | Identifier) -> bool:
     def load_view(self, identifier: str | Identifier) -> View:
         raise NotImplementedError
 
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+        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)

pyiceberg/catalog/glue.py

@@ -1016,6 +1016,10 @@ def view_exists(self, identifier: str | Identifier) -> bool:
     def load_view(self, identifier: str | Identifier) -> View:
         raise NotImplementedError
 
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+        raise NotImplementedError
+
     @staticmethod
     def __is_iceberg_table(table: "TableTypeDef") -> bool:
         return table.get("Parameters", {}).get(TABLE_TYPE, "").lower() == ICEBERG

pyiceberg/catalog/hive.py

@@ -506,6 +506,10 @@ def view_exists(self, identifier: str | Identifier) -> bool:
     def load_view(self, identifier: str | Identifier) -> View:
         raise NotImplementedError
 
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+        raise NotImplementedError
+
     def _create_lock_request(self, database_name: str, table_name: str) -> LockRequest:
         # Iceberg commits are not executed within a Hive transaction, so the lock component uses operationType=NO_TXN.
         # Setting it explicitly also matters for Hive 2.1.0, which rejects a lock component left at the default UNSET

pyiceberg/catalog/noop.py

@@ -175,3 +175,7 @@ def create_view(
     @override
     def load_view(self, identifier: str | Identifier) -> View:
         raise NotImplementedError
+
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+        raise NotImplementedError

pyiceberg/catalog/rest/__init__.py

@@ -176,6 +176,7 @@ class Endpoints:
     register_view: str = "namespaces/{namespace}/register-view"
     drop_view: str = "namespaces/{namespace}/views/{view}"
     view_exists: str = "namespaces/{namespace}/views/{view}"
+    rename_view: str = "views/rename"
     plan_table_scan: str = "namespaces/{namespace}/tables/{table}/plan"
     # Use plan_id (underscore) for str.format; Capability paths use {plan-id} to match the REST spec.
     fetch_planning_result: str = "namespaces/{namespace}/tables/{table}/plan/{plan_id}"
@@ -211,6 +212,7 @@ class Capability:
     V1_CREATE_VIEW = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.create_view}")
     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_RENAME_VIEW = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.rename_view}")
     V1_SUBMIT_TABLE_SCAN_PLAN = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.plan_table_scan}")
     # Spec advertises {plan-id}; must match ConfigResponse endpoint strings from servers.
     V1_FETCH_TABLE_SCAN_PLAN = Endpoint(
@@ -248,6 +250,7 @@ class Capability:
         Capability.V1_LOAD_VIEW,
         Capability.V1_CREATE_VIEW,
         Capability.V1_DELETE_VIEW,
+        Capability.V1_RENAME_VIEW,
     )
 )
 
@@ -1856,6 +1859,29 @@ def drop_view(self, identifier: str | Identifier) -> None:
         except HTTPError as exc:
             _handle_non_200_response(exc, {404: NoSuchViewError})
 
+    @retry(**_RETRY_ARGS)
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+        self._check_endpoint(Capability.V1_RENAME_VIEW)
+        source = self._split_identifier_for_json(from_identifier)
+        destination = self._split_identifier_for_json(to_identifier)
+
+        # Ensure that namespaces exist on source and destination.
+        source_namespace = source["namespace"]
+        if not self.namespace_exists(source_namespace):
+            raise NoSuchNamespaceError(f"Source namespace does not exist: {source_namespace}")
+
+        destination_namespace = destination["namespace"]
+        if not self.namespace_exists(destination_namespace):
+            raise NoSuchNamespaceError(f"Destination namespace does not exist: {destination_namespace}")
+
+        payload = {"source": source, "destination": destination}
+        response = self._session.post(self.url(Endpoints.rename_view), json=payload)
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            _handle_non_200_response(exc, {404: NoSuchViewError, 409: ViewAlreadyExistsError})
+
     def close(self) -> None:
         """Close the catalog and release Session connection adapters.
 

pyiceberg/catalog/sql.py

@@ -837,6 +837,10 @@ def drop_view(self, identifier: str | Identifier) -> None:
     def load_view(self, identifier: str | Identifier) -> View:
         raise NotImplementedError
 
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+        raise NotImplementedError
+
     def close(self) -> None:
         """Close the catalog and release database connections.
 

tests/catalog/test_rest.py

@@ -121,6 +121,7 @@
     Capability.V1_CREATE_VIEW,
     Capability.V1_REGISTER_VIEW,
     Capability.V1_DELETE_VIEW,
+    Capability.V1_RENAME_VIEW,
     Capability.V1_SUBMIT_TABLE_SCAN_PLAN,
     Capability.V1_TABLE_SCAN_PLAN_TASKS,
 ]
@@ -2950,6 +2951,125 @@ def test_drop_view_204(rest_mock: Mocker) -> None:
     RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN).drop_view(("some_namespace", "some_view"))
 
 
+def test_rename_view_204(rest_mock: Mocker) -> None:
+    from_identifier = ("some_namespace", "old_view")
+    to_identifier = ("some_namespace", "new_view")
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/some_namespace",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.post(
+        f"{TEST_URI}v1/views/rename",
+        json={
+            "source": {"namespace": ["some_namespace"], "name": "old_view"},
+            "destination": {"namespace": ["some_namespace"], "name": "new_view"},
+        },
+        status_code=204,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    catalog.rename_view(from_identifier, to_identifier)
+    assert (
+        rest_mock.last_request.text == """{"source": {"namespace": ["some_namespace"], "name": "old_view"}, """
+        """"destination": {"namespace": ["some_namespace"], "name": "new_view"}}"""
+    )
+
+
+def test_rename_view_404(rest_mock: Mocker) -> None:
+    from_identifier = ("some_namespace", "non_existent_view")
+    to_identifier = ("some_namespace", "new_view")
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/some_namespace",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.post(
+        f"{TEST_URI}v1/views/rename",
+        json={
+            "error": {
+                "message": "View does not exist: some_namespace.non_existent_view",
+                "type": "NoSuchViewException",
+                "code": 404,
+            }
+        },
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    with pytest.raises(NoSuchViewError) as exc_info:
+        catalog.rename_view(from_identifier, to_identifier)
+    assert "View does not exist: some_namespace.non_existent_view" in str(exc_info.value)
+
+
+def test_rename_view_409(rest_mock: Mocker) -> None:
+    from_identifier = ("some_namespace", "old_view")
+    to_identifier = ("some_namespace", "existing_view")
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/some_namespace",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.post(
+        f"{TEST_URI}v1/views/rename",
+        json={
+            "error": {
+                "message": "View already exists: some_namespace.existing_view",
+                "type": "ViewAlreadyExistsException",
+                "code": 409,
+            }
+        },
+        status_code=409,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    with pytest.raises(ViewAlreadyExistsError) as exc_info:
+        catalog.rename_view(from_identifier, to_identifier)
+    assert "View already exists: some_namespace.existing_view" in str(exc_info.value)
+
+
+def test_rename_view_source_namespace_does_not_exist(rest_mock: Mocker) -> None:
+    from_identifier = ("non_existent_namespace", "old_view")
+    to_identifier = ("some_namespace", "new_view")
+
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/non_existent_namespace",
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/some_namespace",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    with pytest.raises(NoSuchNamespaceError) as exc_info:
+        catalog.rename_view(from_identifier, to_identifier)
+    assert "Source namespace does not exist: ('non_existent_namespace',)" in str(exc_info.value)
+
+
+def test_rename_view_destination_namespace_does_not_exist(rest_mock: Mocker) -> None:
+    from_identifier = ("some_namespace", "old_view")
+    to_identifier = ("non_existent_namespace", "new_view")
+
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/some_namespace",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/non_existent_namespace",
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    with pytest.raises(NoSuchNamespaceError) as exc_info:
+        catalog.rename_view(from_identifier, to_identifier)
+    assert "Destination namespace does not exist: ('non_existent_namespace',)" in str(exc_info.value)
+
+
 @mock.patch("google.auth.transport.requests.Request")
 @mock.patch("google.auth.load_credentials_from_file")
 def test_rest_catalog_with_google_credentials_path(
@@ -3311,6 +3431,7 @@ def test_view_endpoints_enabled_with_config(self, requests_mock: Mocker) -> None
         # View endpoints should be supported when enabled
         catalog._check_endpoint(Capability.V1_LIST_VIEWS)
         catalog._check_endpoint(Capability.V1_DELETE_VIEW)
+        catalog._check_endpoint(Capability.V1_RENAME_VIEW)
 
 
 def test_table_uuid_check_on_commit(rest_mock: Mocker, example_table_metadata_v2: dict[str, Any]) -> None:

tests/integration/test_catalog.py

@@ -731,6 +731,7 @@ def test_rest_identifiers_with_special_characters(
         table_identifier = (*namespace, f"table {name}")
         view_identifier = (*namespace, f"view {name}")
         registered_view_identifier = (*namespace, f"registered view {name}")
+        renamed_view_identifier = (*namespace, f"renamed view {name}")
 
         rest_catalog.create_namespace(parent)
         rest_catalog.create_namespace(namespace, properties={"key": name})
@@ -758,14 +759,57 @@ def test_rest_identifiers_with_special_characters(
         assert rest_catalog.view_exists(registered_view_identifier)
         assert sorted(rest_catalog.list_views(namespace)) == sorted([view_identifier, registered_view_identifier])
 
-        rest_catalog.drop_view(registered_view_identifier)
+        rest_catalog.rename_view(registered_view_identifier, renamed_view_identifier)
+        assert sorted(rest_catalog.list_views(namespace)) == sorted([view_identifier, renamed_view_identifier])
+
+        rest_catalog.drop_view(renamed_view_identifier)
         rest_catalog.drop_view(view_identifier)
         rest_catalog.drop_table(table_identifier)
         rest_catalog.drop_namespace(namespace)
         rest_catalog.drop_namespace(parent)
         assert not rest_catalog.namespace_exists(parent)
 
 
+@pytest.mark.integration
+def test_rest_rename_view(
+    rest_catalog: RestCatalog, example_view_metadata_v1: dict[str, Any], database_name: str, view_name: str
+) -> None:
+    from_identifier = (database_name, view_name)
+    to_identifier = (database_name, f"{view_name}_renamed")
+
+    rest_catalog.create_namespace_if_not_exists(database_name)
+    view = View(from_identifier, ViewMetadata.model_validate(example_view_metadata_v1))
+
+    rest_catalog.create_view(from_identifier, view.schema(), view.current_version())
+    assert rest_catalog.view_exists(from_identifier)
+
+    rest_catalog.rename_view(from_identifier, to_identifier)
+
+    assert not rest_catalog.view_exists(from_identifier)
+    assert rest_catalog.view_exists(to_identifier)
+
+
+@pytest.mark.integration
+def test_rest_rename_view_across_namespaces(
+    rest_catalog: RestCatalog, example_view_metadata_v1: dict[str, Any], database_name: str, view_name: str
+) -> None:
+    to_database_name = f"{database_name}_renamed"
+    from_identifier = (database_name, view_name)
+    to_identifier = (to_database_name, view_name)
+
+    rest_catalog.create_namespace_if_not_exists(database_name)
+    rest_catalog.create_namespace_if_not_exists(to_database_name)
+    view = View(from_identifier, ViewMetadata.model_validate(example_view_metadata_v1))
+
+    rest_catalog.create_view(from_identifier, view.schema(), view.current_version())
+    assert rest_catalog.view_exists(from_identifier)
+
+    rest_catalog.rename_view(from_identifier, to_identifier)
+
+    assert not rest_catalog.view_exists(from_identifier)
+    assert rest_catalog.view_exists(to_identifier)
+
+
 @pytest.mark.integration
 def test_rest_custom_namespace_separator(rest_catalog: RestCatalog, table_schema_simple: Schema) -> None:
     """