Implement rename view to the rest catalog

pyiceberg/catalog/__init__.py

@@ -744,6 +744,22 @@ 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) -> View:
+        """Rename a fully classified view name.
+
+        Args:
+            from_identifier (str | Identifier): Existing view identifier.
+            to_identifier (str | Identifier): New view identifier.
+
+        Returns:
+            view: the updated view instance with its metadata.
+
+        Raises:
+            NoSuchTableError: If a view with the name does not exist.
+            TableAlreadyExistsError: If the target view already exists.
+        """
+
     @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) -> View:
+        raise NotImplementedError
+
     @override
     def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
         dataset_name = self.identifier_to_database(namespace)

pyiceberg/catalog/dynamodb.py

@@ -580,6 +580,10 @@ def drop_view(self, identifier: str | Identifier) -> None:
     def view_exists(self, identifier: str | Identifier) -> bool:
         raise NotImplementedError
 
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> View:
+        raise NotImplementedError
+
     @override
     def load_view(self, identifier: str | Identifier) -> View:
         raise NotImplementedError

pyiceberg/catalog/glue.py

@@ -1001,6 +1001,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) -> View:
+        raise NotImplementedError
+
     @staticmethod
     def __is_iceberg_table(table: "TableTypeDef") -> bool:
         return table.get("Parameters", {}).get(TABLE_TYPE, "").lower() == ICEBERG

pyiceberg/catalog/hive.py

@@ -874,6 +874,10 @@ def register_view(self, identifier: str | Identifier, metadata_location: str) ->
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> View:
+        raise NotImplementedError
+
     def _get_default_warehouse_location(self, database_name: str, table_name: str) -> str:
         """Override the default warehouse location to follow Hive-style conventions."""
         return self._get_hive_style_warehouse_location(database_name, table_name)

pyiceberg/catalog/noop.py

@@ -161,6 +161,10 @@ def register_view(self, identifier: str | Identifier, metadata_location: str) ->
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 
+    @override
+    def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> View:
+        raise NotImplementedError
+
     @override
     def create_view(
         self,

pyiceberg/catalog/rest/__init__.py

@@ -159,6 +159,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"
     fetch_scan_tasks: str = "namespaces/{namespace}/tables/{table}/tasks"
 
@@ -189,6 +190,7 @@ class Capability:
     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_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}")
     V1_TABLE_SCAN_PLAN_TASKS = Endpoint(http_method=HttpMethod.POST, path=f"{API_PREFIX}/{Endpoints.fetch_scan_tasks}")
 
@@ -218,6 +220,7 @@ class Capability:
         Capability.V1_LIST_VIEWS,
         Capability.V1_LOAD_VIEW,
         Capability.V1_DELETE_VIEW,
+        Capability.V1_RENAME_VIEW,
     )
 )
 
@@ -1503,6 +1506,36 @@ 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) -> View:
+        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})
+
+        return self.load_view(to_identifier)
+
     def close(self) -> None:
         """Close the catalog and release Session connection adapters.
 

pyiceberg/catalog/sql.py

@@ -774,6 +774,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) -> View:
+        raise NotImplementedError
+
     def close(self) -> None:
         """Close the catalog and release database connections.
 

tests/catalog/test_rest.py

@@ -110,6 +110,7 @@
     Capability.V1_VIEW_EXISTS,
     Capability.V1_REGISTER_VIEW,
     Capability.V1_DELETE_VIEW,
+    Capability.V1_RENAME_VIEW,
     Capability.V1_SUBMIT_TABLE_SCAN_PLAN,
     Capability.V1_TABLE_SCAN_PLAN_TASKS,
 ]
@@ -2637,6 +2638,121 @@ 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_200(rest_mock: Mocker, example_view_metadata_rest_json: dict[str, Any]) -> None:
+    rest_mock.post(
+        f"{TEST_URI}v1/views/rename",
+        json={
+            "source": {"namespace": ("pdames",), "name": "source"},
+            "destination": {"namespace": ("pdames",), "name": "destination"},
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/pdames/views/destination",
+        json=example_view_metadata_rest_json,
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/pdames",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    from_identifier = ("pdames", "source")
+    to_identifier = ("pdames", "destination")
+    actual = catalog.rename_view(from_identifier, to_identifier)
+    expected = View(
+        identifier=("pdames", "destination"),
+        metadata=ViewMetadata(**example_view_metadata_rest_json["metadata"]),
+    )
+    assert actual.metadata.model_dump() == expected.metadata.model_dump()
+    assert actual == expected
+
+
+def test_rename_view_from_self_identifier_200(rest_mock: Mocker, example_view_metadata_rest_json: dict[str, Any]) -> None:
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/pdames/views/source",
+        json=example_view_metadata_rest_json,
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    from_identifier = ("pdames", "source")
+    to_identifier = ("pdames", "destination")
+    table = catalog.load_view(from_identifier)
+    rest_mock.post(
+        f"{TEST_URI}v1/views/rename",
+        json={
+            "source": {"namespace": ("pdames",), "name": "source"},
+            "destination": {"namespace": ("pdames",), "name": "destination"},
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/pdames/views/destination",
+        json=example_view_metadata_rest_json,
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/pdames",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    actual = catalog.rename_view(table.name(), to_identifier)
+    expected = View(
+        identifier=("pdames", "destination"),
+        metadata=ViewMetadata(**example_view_metadata_rest_json["metadata"]),
+    )
+    assert actual.metadata.model_dump() == expected.metadata.model_dump()
+    assert actual == expected
+
+
+def test_rename_view_source_namespace_does_not_exist(rest_mock: Mocker) -> None:
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    from_identifier = ("invalid", "source")
+    to_identifier = ("pdames", "destination")
+
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/invalid",
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/pdames",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+
+    with pytest.raises(NoSuchNamespaceError) as e:
+        catalog.rename_view(from_identifier, to_identifier)
+    assert "Source namespace does not exist" in str(e.value)
+
+
+def test_rename_view_destination_namespace_does_not_exist(rest_mock: Mocker) -> None:
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    from_identifier = ("pdames", "source")
+    to_identifier = ("invalid", "destination")
+
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/pdames",
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    rest_mock.head(
+        f"{TEST_URI}v1/namespaces/invalid",
+        status_code=404,
+        request_headers=TEST_HEADERS,
+    )
+
+    with pytest.raises(NoSuchNamespaceError) as e:
+        catalog.rename_view(from_identifier, to_identifier)
+    assert "Destination namespace does not exist" in str(e.value)
+
+
 @mock.patch("google.auth.transport.requests.Request")
 @mock.patch("google.auth.load_credentials_from_file")
 def test_rest_catalog_with_google_credentials_path(