Add rename_view to REST Catalog

pyiceberg/catalog/__init__.py

@@ -744,6 +744,19 @@ 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 classified 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 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

@@ -334,6 +334,9 @@ 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 load_namespace_properties(self, namespace: str | Identifier) -> Properties:
  dataset_name = self.identifier_to_database(namespace)
 

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

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

pyiceberg/catalog/hive.py

@@ -490,6 +490,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:
  lock_component: LockComponent = LockComponent(
  level=LockLevel.TABLE, type=LockType.EXCLUSIVE, dbname=database_name, tablename=table_name, isTransactional=True

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

@@ -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,29 @@ def drop_view(self, identifier: str | Identifier) -> None:
  except HTTPError as exc:
  _handle_non_200_response(exc, {404: NoSuchViewError})
 
+ @retry(**_RETRY_ARGS)
+ def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+ self._check_endpoint(Capability.V1_RENAME_VIEW)
+ payload = {
+ "source": self._split_identifier_for_json(from_identifier),
+ "destination": self._split_identifier_for_json(to_identifier),
+ }
+
+ # Ensure source and destination namespaces exist before rename.
+ source_namespace = self._split_identifier_for_json(from_identifier)["namespace"]
+ dest_namespace = self._split_identifier_for_json(to_identifier)["namespace"]
+
+ if not self.namespace_exists(source_namespace):
+ raise NoSuchNamespaceError(f"Source namespace does not exist: {source_namespace}")
+ if not self.namespace_exists(dest_namespace):
+ raise NoSuchNamespaceError(f"Destination namespace does not exist: {dest_namespace}")
+
+ 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

@@ -784,3 +784,6 @@ def close(self) -> None:
  """
  if hasattr(self, "engine"):
  self.engine.dispose()
+
+ def rename_view(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> None:
+ raise NotImplementedError

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,
 ]
@@ -3196,3 +3197,122 @@ def test_load_table_without_storage_credentials(
  )
  assert actual.metadata.model_dump() == expected.metadata.model_dump()
  assert actual == expected
+
+
+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)

tests/integration/test_catalog.py

@@ -670,6 +670,26 @@ def test_rest_drop_view(
 
 
 @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
+@pytest.mark.skip(reason="Requires Iceberg REST Fixtures 1.11.x")
 def test_rest_custom_namespace_separator(rest_catalog: RestCatalog, table_schema_simple: Schema) -> None:
  """
  Tests that the REST catalog correctly picks up the namespace-separator from the config endpoint.