REST: Don't URL-encode view names in create_view and register_view bodies

pyiceberg/catalog/rest/__init__.py

@@ -1312,7 +1312,7 @@ def create_view(
             location = location.rstrip("/")
 
         request = CreateViewRequest(
-            name=namespace_and_view["view"],
+            name=self._identifier_to_validated_tuple(identifier)[-1],
             location=location,
             view_schema=fresh_schema,
             view_version=view_version,
@@ -1826,15 +1826,14 @@ def view_exists(self, identifier: str | Identifier) -> bool:
     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"]
+        identifier_tuple = self._identifier_to_validated_tuple(identifier)
         if self.table_exists(identifier):
-            raise TableAlreadyExistsError(f"Table {namespace}.{view} already exists")
+            raise TableAlreadyExistsError(f"Table {'.'.join(identifier_tuple)} already exists")
 
-        request = RegisterViewRequest(name=view, metadata_location=metadata_location)
+        request = RegisterViewRequest(name=identifier_tuple[-1], metadata_location=metadata_location)
         serialized_json = request.model_dump_json().encode(UTF8)
         response = self._session.post(
-            self.url(Endpoints.register_view, namespace=namespace),
+            self.url(Endpoints.register_view, namespace=namespace_and_view["namespace"]),
             data=serialized_json,
         )
         try:

tests/catalog/test_rest.py

@@ -23,7 +23,7 @@
 from contextlib import contextmanager
 from typing import Any, cast
 from unittest import mock
-from urllib.parse import parse_qs, urlsplit
+from urllib.parse import parse_qs, quote, urlsplit
 
 import pytest
 from requests import Request
@@ -1805,6 +1805,27 @@ def test_create_view_200(rest_mock: Mocker, table_schema_simple: Schema, example
     assert actual.config == example_view_metadata_rest_json["config"]
 
 
+def test_create_view_name_is_not_encoded_in_body(
+    rest_mock: Mocker, table_schema_simple: Schema, example_view_metadata_rest_json: dict[str, Any]
+) -> None:
+    rest_mock.post(
+        f"{TEST_URI}v1/namespaces/fokko/views",
+        json=example_view_metadata_rest_json,
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+    view_version = ViewVersion(
+        schema_id=1,
+        summary={"engine-name": "spark", "engineVersion": "3.3"},
+        representations=[],
+        default_namespace=["default"],
+    )
+    for view_name in ["my view", "dept/view", "100%", "\u65e5\u672c"]:
+        catalog.create_view(identifier=("fokko", view_name), schema=table_schema_simple, view_version=view_version)
+        assert rest_mock.last_request.json()["name"] == view_name
+
+
 def test_create_view_409(
     rest_mock: Mocker,
     table_schema_simple: Schema,
@@ -2817,6 +2838,24 @@ def test_register_view_200(rest_mock: Mocker, example_view_metadata_rest_json: d
     assert actual.config == example_view_metadata_rest_json["config"]
 
 
+def test_register_view_name_is_not_encoded_in_body(rest_mock: Mocker, example_view_metadata_rest_json: dict[str, Any]) -> None:
+    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)
+    for view_name in ["my view", "dept/view", "100%", "\u65e5\u672c"]:
+        rest_mock.head(
+            f"{TEST_URI}v1/namespaces/default/tables/{quote(view_name, safe='')}",
+            status_code=404,
+            request_headers=TEST_HEADERS,
+        )
+        catalog.register_view(identifier=("default", view_name), metadata_location="s3://warehouse/database/view/metadata.json")
+        assert rest_mock.last_request.json()["name"] == view_name
+
+
 def test_register_view_409_view(rest_mock: Mocker) -> None:
     rest_mock.head(
         f"{TEST_URI}v1/namespaces/default/tables/registered_view",

tests/integration/test_catalog.py

@@ -719,6 +719,53 @@ def test_rest_drop_view(
     assert not rest_catalog.view_exists(identifier)
 
 
+@pytest.mark.integration
+def test_rest_identifiers_with_special_characters(
+    rest_catalog: RestCatalog, table_schema_simple: Schema, example_view_metadata_v1: dict[str, Any]
+) -> None:
+    """Names that need URL encoding must round-trip unchanged through path, query, and body parameters."""
+    # "%" is omitted because the REST fixture's Jetty rejects "%25" in URL paths ("400 Ambiguous URI path encoding")
+    for name in ["with space", "a+b&c=d", "\u65e5\u672c"]:
+        parent = (f"{name}_{uuid.uuid4().hex}",)
+        namespace = (*parent, name)
+        table_identifier = (*namespace, f"table {name}")
+        view_identifier = (*namespace, f"view {name}")
+        registered_view_identifier = (*namespace, f"registered view {name}")
+
+        rest_catalog.create_namespace(parent)
+        rest_catalog.create_namespace(namespace, properties={"key": name})
+        assert parent in rest_catalog.list_namespaces()
+        assert rest_catalog.list_namespaces(parent) == [namespace]
+        assert rest_catalog.namespace_exists(namespace)
+        assert rest_catalog.load_namespace_properties(namespace)["key"] == name
+
+        rest_catalog.create_table(table_identifier, table_schema_simple)
+        assert rest_catalog.table_exists(table_identifier)
+        assert rest_catalog.list_tables(namespace) == [table_identifier]
+        table = rest_catalog.load_table(table_identifier)
+        assert table.name() == table_identifier
+        with table.transaction() as tx:
+            tx.set_properties(key=name)
+        assert rest_catalog.load_table(table_identifier).properties["key"] == name
+
+        view = View(view_identifier, ViewMetadata.model_validate(example_view_metadata_v1))
+        created_view = rest_catalog.create_view(view_identifier, view.schema(), view.current_version())
+        assert rest_catalog.view_exists(view_identifier)
+        assert rest_catalog.list_views(namespace) == [view_identifier]
+
+        assert created_view.metadata_location is not None
+        rest_catalog.register_view(registered_view_identifier, created_view.metadata_location)
+        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.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_custom_namespace_separator(rest_catalog: RestCatalog, table_schema_simple: Schema) -> None:
     """