Fix issues related to having catalog_name in identifier

pyiceberg/catalog/rest.py

@@ -354,7 +354,10 @@ def _identifier_to_validated_tuple(self, identifier: Union[str, Identifier]) ->
 
     def _split_identifier_for_path(self, identifier: Union[str, Identifier, TableIdentifier]) -> Properties:
         if isinstance(identifier, TableIdentifier):
-            return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root[1:]), "table": identifier.name}
+            if identifier.namespace.root[0] == self.name:
+                return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root[1:]), "table": identifier.name}
+            else:
+                return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root), "table": identifier.name}
         identifier_tuple = self._identifier_to_validated_tuple(identifier)
         return {"namespace": NAMESPACE_SEPARATOR.join(identifier_tuple[:-1]), "table": identifier_tuple[-1]}
 
@@ -675,6 +678,17 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
 
         return self.load_table(to_identifier)
 
+    def _remove_catalog_name_from_table_request_identifier(self, table_request: CommitTableRequest) -> CommitTableRequest:
+        if table_request.identifier.namespace.root[0] == self.name:
+            return table_request.model_copy(
+                update={
+                    "identifier": TableIdentifier(
+                        namespace=table_request.identifier.namespace.root[1:], name=table_request.identifier.name
+                    ).model_dump()
+                }
+            )
+        return table_request
+
     @retry(**_RETRY_ARGS)
     def _commit_table(self, table_request: CommitTableRequest) -> CommitTableResponse:
         """Update the table.
@@ -692,7 +706,7 @@ def _commit_table(self, table_request: CommitTableRequest) -> CommitTableRespons
         """
         response = self._session.post(
             self.url(Endpoints.update_table, prefixed=True, **self._split_identifier_for_path(table_request.identifier)),
-            data=table_request.model_dump_json().encode(UTF8),
+            data=self._remove_catalog_name_from_table_request_identifier(table_request).model_dump_json().encode(UTF8),
         )
         try:
             response.raise_for_status()

tests/catalog/test_rest.py

@@ -37,7 +37,7 @@
 from pyiceberg.io import load_file_io
 from pyiceberg.partitioning import PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
-from pyiceberg.table import Table
+from pyiceberg.table import CommitTableRequest, Table, TableIdentifier
 from pyiceberg.table.metadata import TableMetadataV1
 from pyiceberg.table.sorting import SortField, SortOrder
 from pyiceberg.transforms import IdentityTransform, TruncateTransform
@@ -1226,3 +1226,25 @@ def test_catalog_from_parameters_empty_env(rest_mock: Mocker) -> None:
 
     catalog = cast(RestCatalog, load_catalog("production", uri="https://other-service.io/api"))
     assert catalog.uri == "https://other-service.io/api"
+
+
+def test_commit_table_fail(rest_mock: Mocker, example_table_metadata_v2: Dict[str, Any]) -> None:
+    test_table_request = CommitTableRequest(
+        identifier=TableIdentifier(namespace=("catalog_name", "namespace"), name="table_name"),
+        updates=[],
+        requirements=[],
+    )
+    rest_mock.post(
+        url=f"{TEST_URI}v1/namespaces/namespace/tables/table_name",
+        json={
+            "metadata": example_table_metadata_v2,
+            "metadata-location": "test",
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    RestCatalog("catalog_name", uri=TEST_URI, token=TEST_TOKEN)._commit_table(test_table_request)
+    assert (
+        rest_mock.last_request.text
+        == """{"identifier":{"namespace":["namespace"],"name":"table_name"},"requirements":[],"updates":[]}"""
+    )

tests/integration/test_writes/test_rest_behavior.py

@@ -0,0 +1,68 @@
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+# pylint:disable=redefined-outer-name
+from typing import List, Optional, Union
+
+import pyarrow as pa
+import pytest
+
+from pyiceberg.catalog import Catalog, load_catalog
+from pyiceberg.exceptions import NoSuchTableError
+from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
+from pyiceberg.schema import Schema
+from pyiceberg.table import Table
+from pyiceberg.typedef import EMPTY_DICT, Properties
+from tests.integration.test_writes.utils import TABLE_SCHEMA
+
+
+def _create_table(
+    session_catalog: Catalog,
+    identifier: str,
+    properties: Properties = EMPTY_DICT,
+    data: Optional[List[pa.Table]] = None,
+    partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+    schema: Union[Schema, "pa.Schema"] = TABLE_SCHEMA,
+) -> Table:
+    try:
+        session_catalog.drop_table(identifier=identifier)
+    except NoSuchTableError:
+        pass
+
+    tbl = session_catalog.create_table(identifier=identifier, schema=schema, properties=properties, partition_spec=partition_spec)
+
+    if data is not None:
+        for d in data:
+            tbl.append(d)
+
+    return tbl
+
+
+@pytest.mark.integration
+def test_rest_catalog_with_empty_name(arrow_table_with_null: pa.Table) -> None:
+    identifier = "default.test_rest_append"
+    test_catalog = load_catalog(
+        "",  # intentionally empty
+        **{
+            "type": "rest",
+            "uri": "http://localhost:8181",
+            "s3.endpoint": "http://localhost:9000",
+            "s3.access-key-id": "admin",
+            "s3.secret-access-key": "password",
+        },
+    )
+    tbl = _create_table(test_catalog, identifier, data=[])
+    tbl.append(arrow_table_with_null)