Raise on view-table collisions in create/register/rename

- Add `_raise_if_view_exists` helper that maps `view_exists(id) == True` at
  the target identifier to `TableAlreadyExistsError`. Catalogs without view
  support (most non-REST impls) raise `NotImplementedError` from
  `view_exists`; treat that as "no view at this identifier".
- Wire the check into `create_table`, `register_table`, and `rename_table`
  on SqlCatalog, HiveCatalog, GlueCatalog, DynamoDbCatalog, and
  BigQueryMetastoreCatalog (skipping methods that already raise
  `NotImplementedError`). Also call it from
  `MetastoreCatalog.create_table_transaction`.
- For `rename_table`, the check is on the destination identifier.
- RestCatalog intentionally unchanged: the server's 409 response is the
  authority and an extra client-side HEAD per call would be wasted.
- Re-uses `TableAlreadyExistsError` so existing `create_table_if_not_exists`
  callers see the same exception type they already catch.
- Tests: unit tests for the helper (returns True / False / raises
  `NotImplementedError`) in `tests/catalog/test_base.py`, plus
  parametrized behavior tests over InMemoryCatalog + SqlCatalog in
  `tests/catalog/test_catalog_behaviors.py` that monkeypatch `view_exists`
  to assert each call site invokes the helper.

Author: smaheshwar-pltr

pyiceberg/catalog/__init__.py

@@ -323,6 +323,20 @@ def delete_data_files(io: FileIO, manifests_to_delete: list[ManifestFile]) -> No
                 deleted_files[path] = True
 
 
+def _raise_if_view_exists(catalog: Catalog, identifier: str | Identifier) -> None:
+    """Raise `TableAlreadyExistsError` if a view exists at the given identifier.
+
+    Catalogs that don't support views raise `NotImplementedError` from `view_exists` —
+    treat that as "no view at this identifier".
+    """
+    try:
+        view_collision = catalog.view_exists(identifier)
+    except NotImplementedError:
+        view_collision = False
+    if view_collision:
+        raise TableAlreadyExistsError(f"View with same name already exists: {identifier}")
+
+
 def _import_catalog(name: str, catalog_impl: str, properties: Properties) -> Catalog | None:
     try:
         path_parts = catalog_impl.split(".")
@@ -920,6 +934,7 @@ def create_table_transaction(
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
         properties: Properties = EMPTY_DICT,
     ) -> CreateTableTransaction:
+        _raise_if_view_exists(self, identifier)
         return CreateTableTransaction(
             self._create_staged_table(identifier, schema, location, partition_spec, sort_order, properties)
         )

pyiceberg/catalog/bigquery_metastore.py

@@ -28,7 +28,7 @@
 from google.oauth2 import service_account
 from typing_extensions import override
 
-from pyiceberg.catalog import WAREHOUSE_LOCATION, MetastoreCatalog, PropertiesUpdateSummary
+from pyiceberg.catalog import WAREHOUSE_LOCATION, MetastoreCatalog, PropertiesUpdateSummary, _raise_if_view_exists
 from pyiceberg.exceptions import NamespaceAlreadyExistsError, NoSuchNamespaceError, NoSuchTableError, TableAlreadyExistsError
 from pyiceberg.io import load_file_io
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
@@ -134,6 +134,7 @@ def create_table(
         schema: Schema = self._convert_schema_if_needed(schema)  # type: ignore
 
         dataset_name, table_name = self.identifier_to_database_and_table(identifier)
+        _raise_if_view_exists(self, identifier)
 
         location = self._resolve_table_location(location, dataset_name, table_name)
         provider = load_location_provider(table_location=location, table_properties=properties)
@@ -295,6 +296,7 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o
         if overwrite:
             raise NotImplementedError("`overwrite` isn't supported")
 
+        _raise_if_view_exists(self, identifier)
         dataset_name, table_name = self.identifier_to_database_and_table(identifier)
 
         dataset_ref = DatasetReference(project=self.project_id, dataset_id=dataset_name)

pyiceberg/catalog/dynamodb.py

@@ -34,6 +34,7 @@
     TABLE_TYPE,
     MetastoreCatalog,
     PropertiesUpdateSummary,
+    _raise_if_view_exists,
 )
 from pyiceberg.exceptions import (
     ConditionalCheckFailedException,
@@ -187,6 +188,7 @@ def create_table(
         )
 
         database_name, table_name = self.identifier_to_database_and_table(identifier)
+        _raise_if_view_exists(self, identifier)
 
         location = self._resolve_table_location(location, database_name, table_name)
         provider = load_location_provider(table_location=location, table_properties=properties)
@@ -313,6 +315,7 @@ def rename_table(self, from_identifier: str | Identifier, to_identifier: str | I
         """
         from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
+        _raise_if_view_exists(self, to_identifier)
 
         from_table_item = self._get_iceberg_table_item(database_name=from_database_name, table_name=from_table_name)
 

pyiceberg/catalog/glue.py

@@ -39,6 +39,7 @@
     TABLE_TYPE,
     MetastoreCatalog,
     PropertiesUpdateSummary,
+    _raise_if_view_exists,
 )
 from pyiceberg.exceptions import (
     CommitFailedException,
@@ -571,6 +572,7 @@ def create_table(
 
         """
         database_name, table_name = self.identifier_to_database_and_table(identifier)
+        _raise_if_view_exists(self, identifier)
 
         if self._is_s3tables_database(database_name):
             return self._create_table_s3tables(
@@ -621,6 +623,7 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o
         if overwrite:
             raise NotImplementedError("`overwrite` isn't supported")
 
+        _raise_if_view_exists(self, identifier)
         database_name, table_name = self.identifier_to_database_and_table(identifier)
         properties = EMPTY_DICT
         io = self._load_file_io(location=metadata_location)
@@ -772,6 +775,7 @@ def rename_table(self, from_identifier: str | Identifier, to_identifier: str | I
         """
         from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
+        _raise_if_view_exists(self, to_identifier)
         try:
             get_table_response = self.glue.get_table(DatabaseName=from_database_name, Name=from_table_name)
         except self.glue.exceptions.EntityNotFoundException as e:

pyiceberg/catalog/hive.py

@@ -61,6 +61,7 @@
     URI,
     MetastoreCatalog,
     PropertiesUpdateSummary,
+    _raise_if_view_exists,
 )
 from pyiceberg.exceptions import (
     CommitFailedException,
@@ -413,6 +414,7 @@ def create_table(
             ValueError: If the identifier is invalid.
         """
         properties = {**DEFAULT_PROPERTIES, **properties}
+        _raise_if_view_exists(self, identifier)
         staged_table = self._create_staged_table(
             identifier=identifier,
             schema=schema,
@@ -461,6 +463,7 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o
         if overwrite:
             raise NotImplementedError("`overwrite` isn't supported")
 
+        _raise_if_view_exists(self, identifier)
         database_name, table_name = self.identifier_to_database_and_table(identifier)
         io = self._load_file_io(location=metadata_location)
         metadata_file = io.new_input(metadata_location)
@@ -700,6 +703,7 @@ def rename_table(self, from_identifier: str | Identifier, to_identifier: str | I
 
         if self.table_exists(to_identifier):
             raise TableAlreadyExistsError(f"Table already exists: {to_table_name}")
+        _raise_if_view_exists(self, to_identifier)
 
         try:
             with self._client as open_client:

pyiceberg/catalog/sql.py

@@ -46,6 +46,7 @@
     Catalog,
     MetastoreCatalog,
     PropertiesUpdateSummary,
+    _raise_if_view_exists,
 )
 from pyiceberg.exceptions import (
     CommitFailedException,
@@ -211,6 +212,7 @@ def create_table(
         table_name = Catalog.table_name_from(identifier)
         if not self.namespace_exists(namespace_identifier):
             raise NoSuchNamespaceError(f"Namespace does not exist: {namespace_identifier}")
+        _raise_if_view_exists(self, identifier)
 
         namespace = Catalog.namespace_to_string(namespace_identifier)
         location = self._resolve_table_location(location, namespace, table_name)
@@ -263,6 +265,7 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o
         table_name = Catalog.table_name_from(identifier)
         if not self.namespace_exists(namespace):
             raise NoSuchNamespaceError(f"Namespace does not exist: {namespace}")
+        _raise_if_view_exists(self, identifier)
 
         with Session(self.engine) as session:
             try:
@@ -376,6 +379,7 @@ def rename_table(self, from_identifier: str | Identifier, to_identifier: str | I
         to_table_name = Catalog.table_name_from(to_identifier)
         if not self.namespace_exists(to_namespace):
             raise NoSuchNamespaceError(f"Namespace does not exist: {to_namespace}")
+        _raise_if_view_exists(self, to_identifier)
         with Session(self.engine) as session:
             try:
                 if self.engine.dialect.supports_sane_rowcount:

tests/catalog/test_base.py

@@ -22,8 +22,9 @@
 
 import pytest
 
-from pyiceberg.catalog import Catalog, load_catalog
+from pyiceberg.catalog import Catalog, _raise_if_view_exists, load_catalog
 from pyiceberg.catalog.memory import InMemoryCatalog
+from pyiceberg.exceptions import TableAlreadyExistsError
 from pyiceberg.io import WAREHOUSE
 from pyiceberg.schema import Schema
 from pyiceberg.types import NestedField, StringType
@@ -69,6 +70,38 @@ def test_catalog_repr(catalog: InMemoryCatalog) -> None:
     assert s == "test.in_memory.catalog (<class 'pyiceberg.catalog.memory.InMemoryCatalog'>)"
 
 
+class _StubCatalog:
+    def __init__(self, *, returns: bool | None = None, raises: type[Exception] | None = None) -> None:
+        self._returns = returns
+        self._raises = raises
+        self.calls: list[object] = []
+
+    def view_exists(self, identifier: object) -> bool:
+        self.calls.append(identifier)
+        if self._raises is not None:
+            raise self._raises
+        assert self._returns is not None
+        return self._returns
+
+
+class TestRaiseIfViewExists:
+    def test_raises_when_view_exists(self) -> None:
+        stub = _StubCatalog(returns=True)
+        with pytest.raises(TableAlreadyExistsError, match="View with same name already exists: ns.t"):
+            _raise_if_view_exists(stub, "ns.t")  # type: ignore[arg-type]
+        assert stub.calls == ["ns.t"]
+
+    def test_no_raise_when_view_absent(self) -> None:
+        stub = _StubCatalog(returns=False)
+        _raise_if_view_exists(stub, ("ns", "t"))  # type: ignore[arg-type]
+        assert stub.calls == [("ns", "t")]
+
+    def test_not_implemented_treated_as_no_view(self) -> None:
+        stub = _StubCatalog(raises=NotImplementedError)
+        _raise_if_view_exists(stub, "ns.t")  # type: ignore[arg-type]
+        assert stub.calls == ["ns.t"]
+
+
 class TestCatalogClose:
     """Test catalog close functionality."""
 

tests/catalog/test_catalog_behaviors.py

@@ -461,6 +461,77 @@ def test_rename_table_to_missing_namespace(
         catalog.rename_table(test_table_identifier, another_table_identifier)
 
 
+# View-collision tests
+#
+# InMemoryCatalog and SqlCatalog don't support views (`view_exists` raises
+# `NotImplementedError`), so we monkeypatch it to assert the call sites invoke
+# the helper that maps a view at the target identifier to `TableAlreadyExistsError`.
+
+
+def test_create_table_raises_when_view_exists_at_identifier(
+    catalog: Catalog,
+    test_table_identifier: Identifier,
+    table_schema_simple: Schema,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    namespace = Catalog.namespace_from(test_table_identifier)
+    catalog.create_namespace(namespace)
+    monkeypatch.setattr(
+        catalog, "view_exists", lambda identifier: Catalog.identifier_to_tuple(identifier) == test_table_identifier
+    )
+    with pytest.raises(TableAlreadyExistsError, match="View with same name already exists"):
+        catalog.create_table(test_table_identifier, table_schema_simple)
+
+
+def test_create_table_transaction_raises_when_view_exists_at_identifier(
+    catalog: Catalog,
+    test_table_identifier: Identifier,
+    table_schema_simple: Schema,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    namespace = Catalog.namespace_from(test_table_identifier)
+    catalog.create_namespace(namespace)
+    monkeypatch.setattr(
+        catalog, "view_exists", lambda identifier: Catalog.identifier_to_tuple(identifier) == test_table_identifier
+    )
+    with pytest.raises(TableAlreadyExistsError, match="View with same name already exists"):
+        catalog.create_table_transaction(test_table_identifier, table_schema_simple)
+
+
+def test_register_table_raises_when_view_exists_at_identifier(
+    catalog: Catalog,
+    test_table_identifier: Identifier,
+    metadata_location: str,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    namespace = Catalog.namespace_from(test_table_identifier)
+    catalog.create_namespace(namespace)
+    monkeypatch.setattr(
+        catalog, "view_exists", lambda identifier: Catalog.identifier_to_tuple(identifier) == test_table_identifier
+    )
+    with pytest.raises(TableAlreadyExistsError, match="View with same name already exists"):
+        catalog.register_table(test_table_identifier, metadata_location)
+
+
+def test_rename_table_raises_when_view_exists_at_destination(
+    catalog: Catalog,
+    table_schema_simple: Schema,
+    test_table_identifier: Identifier,
+    another_table_identifier: Identifier,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from_namespace = Catalog.namespace_from(test_table_identifier)
+    to_namespace = Catalog.namespace_from(another_table_identifier)
+    catalog.create_namespace(from_namespace)
+    catalog.create_namespace(to_namespace)
+    catalog.create_table(test_table_identifier, table_schema_simple)
+    monkeypatch.setattr(
+        catalog, "view_exists", lambda identifier: Catalog.identifier_to_tuple(identifier) == another_table_identifier
+    )
+    with pytest.raises(TableAlreadyExistsError, match="View with same name already exists"):
+        catalog.rename_table(test_table_identifier, another_table_identifier)
+
+
 # Drop table tests