Hive metastore register table

pyiceberg/catalog/hive.py

@@ -404,7 +404,22 @@ def register_table(self, identifier: Union[str, Identifier], metadata_location:
         Raises:
             TableAlreadyExistsError: If the table already exists
         """
-        raise NotImplementedError
+        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)
+        staged_table = StagedTable(
+            identifier=(database_name, table_name),
+            metadata=FromInputFile.table_metadata(metadata_file),
+            metadata_location=metadata_location,
+            io=io,
+            catalog=self,
+        )
+        tbl = self._convert_iceberg_into_hive(staged_table)
+        with self._client as open_client:
+            self._create_hive_table(open_client, tbl)
+            hive_table = open_client.get_table(dbname=database_name, tbl_name=table_name)
+
+        return self._convert_hive_into_iceberg(hive_table)
 
     def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         raise NotImplementedError

tests/integration/test_register_table.py

@@ -0,0 +1,116 @@
+# 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.
+import pytest
+
+from pyiceberg.catalog import Catalog
+from pyiceberg.catalog.hive import (
+    HiveCatalog,
+)
+from pyiceberg.exceptions import NoSuchTableError, TableAlreadyExistsError
+from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
+from pyiceberg.schema import Schema
+from pyiceberg.table import Table
+from pyiceberg.types import (
+    BooleanType,
+    DateType,
+    IntegerType,
+    NestedField,
+    StringType,
+)
+
+TABLE_SCHEMA = Schema(
+    NestedField(field_id=1, name="foo", field_type=BooleanType(), required=False),
+    NestedField(field_id=2, name="bar", field_type=StringType(), required=False),
+    NestedField(field_id=4, name="baz", field_type=IntegerType(), required=False),
+    NestedField(field_id=10, name="qux", field_type=DateType(), required=False),
+)
+
+
+def _create_table(
+    session_catalog: Catalog,
+    identifier: str,
+    format_version: int,
+    location: str,
+    partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
+    schema: Schema = TABLE_SCHEMA,
+) -> Table:
+    try:
+        session_catalog.drop_table(identifier=identifier)
+    except NoSuchTableError:
+        pass
+
+    return session_catalog.create_table(
+        identifier=identifier,
+        schema=schema,
+        location=location,
+        properties={"format-version": str(format_version)},
+        partition_spec=partition_spec,
+    )
+
+
+@pytest.mark.integration
+def test_hive_register_table(
+    session_catalog: HiveCatalog,
+) -> None:
+    identifier = "default.hive_register_table"
+    location = "s3a://warehouse/default/hive_register_table"
+    tbl = _create_table(session_catalog, identifier, 2, location)
+    assert session_catalog.table_exists(identifier=identifier)
+    session_catalog.drop_table(identifier=identifier)
+    assert not session_catalog.table_exists(identifier=identifier)
+    session_catalog.register_table(("default", "hive_register_table"), metadata_location=tbl.metadata_location)
+    assert session_catalog.table_exists(identifier=identifier)
+
+
+@pytest.mark.integration
+def test_hive_register_table_existing(
+    session_catalog: HiveCatalog,
+) -> None:
+    identifier = "default.hive_register_table_existing"
+    location = "s3a://warehouse/default/hive_register_table_existing"
+    tbl = _create_table(session_catalog, identifier, 2, location)
+    assert session_catalog.table_exists(identifier=identifier)
+    # Assert that registering the table again raises TableAlreadyExistsError
+    with pytest.raises(TableAlreadyExistsError):
+        session_catalog.register_table(("default", "hive_register_table_existing"), metadata_location=tbl.metadata_location)
+
+
+@pytest.mark.integration
+def test_rest_register_table(
+    session_catalog: Catalog,
+) -> None:
+    identifier = "default.rest_register_table"
+    location = "s3a://warehouse/default/rest_register_table"
+    tbl = _create_table(session_catalog, identifier, 2, location)
+    assert session_catalog.table_exists(identifier=identifier)
+    session_catalog.drop_table(identifier=identifier)
+    assert not session_catalog.table_exists(identifier=identifier)
+    session_catalog.register_table(identifier=identifier, metadata_location=tbl.metadata_location)
+    assert session_catalog.table_exists(identifier=identifier)
+
+
+@pytest.mark.integration
+def test_rest_register_table_existing(
+    session_catalog: Catalog,
+) -> None:
+    identifier = "default.rest_register_table_existing"
+    location = "s3a://warehouse/default/rest_register_table_existing"
+    tbl = _create_table(session_catalog, identifier, 2, location)
+    assert session_catalog.table_exists(identifier=identifier)
+    # Assert that registering the table again raises TableAlreadyExistsError
+    with pytest.raises(TableAlreadyExistsError):
+        session_catalog.register_table(identifier=identifier, metadata_location=tbl.metadata_location)