You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by bl...@apache.org on 2023/09/04 16:00:50 UTC
[iceberg] branch master updated: Python: Add register table (#8465)
This is an automated email from the ASF dual-hosted git repository.
blue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new be818b65f3 Python: Add register table (#8465)
be818b65f3 is described below
commit be818b65f36250cf36a20719b31baf3dc3ca076e
Author: Fokko Driesprong <fo...@apache.org>
AuthorDate: Mon Sep 4 18:00:44 2023 +0200
Python: Add register table (#8465)
---
python/pyiceberg/catalog/__init__.py | 15 ++++
python/pyiceberg/catalog/dynamodb.py | 15 ++++
python/pyiceberg/catalog/glue.py | 15 ++++
python/pyiceberg/catalog/hive.py | 15 ++++
python/pyiceberg/catalog/noop.py | 15 ++++
python/pyiceberg/catalog/rest.py | 37 ++++++++++
python/pyiceberg/catalog/sql.py | 15 ++++
python/tests/catalog/test_base.py | 3 +
python/tests/catalog/test_rest.py | 136 +++++++++++++++++++++++++++++++++++
9 files changed, 266 insertions(+)
diff --git a/python/pyiceberg/catalog/__init__.py b/python/pyiceberg/catalog/__init__.py
index 7d42527a1c..dafef71287 100644
--- a/python/pyiceberg/catalog/__init__.py
+++ b/python/pyiceberg/catalog/__init__.py
@@ -314,6 +314,21 @@ class Catalog(ABC):
NoSuchTableError: If a table with the name does not exist.
"""
+ @abstractmethod
+ def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
+ """Register a new table using existing metadata.
+
+ Args:
+ identifier Union[str, Identifier]: Table identifier for the table
+ metadata_location str: The location to the metadata
+
+ Returns:
+ Table: The newly registered table
+
+ Raises:
+ TableAlreadyExistsError: If the table already exists
+ """
+
@abstractmethod
def drop_table(self, identifier: Union[str, Identifier]) -> None:
"""Drop a table.
diff --git a/python/pyiceberg/catalog/dynamodb.py b/python/pyiceberg/catalog/dynamodb.py
index bf5dc8067a..848ec03126 100644
--- a/python/pyiceberg/catalog/dynamodb.py
+++ b/python/pyiceberg/catalog/dynamodb.py
@@ -168,6 +168,21 @@ class DynamoDbCatalog(Catalog):
return self.load_table(identifier=identifier)
+ def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
+ """Register a new table using existing metadata.
+
+ Args:
+ identifier Union[str, Identifier]: Table identifier for the table
+ metadata_location str: The location to the metadata
+
+ Returns:
+ Table: The newly registered table
+
+ Raises:
+ TableAlreadyExistsError: If the table already exists
+ """
+ raise NotImplementedError
+
def _commit_table(self, table_request: CommitTableRequest) -> CommitTableResponse:
"""Update the table.
diff --git a/python/pyiceberg/catalog/glue.py b/python/pyiceberg/catalog/glue.py
index a191cfe25a..405df6720f 100644
--- a/python/pyiceberg/catalog/glue.py
+++ b/python/pyiceberg/catalog/glue.py
@@ -225,6 +225,21 @@ class GlueCatalog(Catalog):
return self.load_table(identifier=identifier)
+ def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
+ """Register a new table using existing metadata.
+
+ Args:
+ identifier Union[str, Identifier]: Table identifier for the table
+ metadata_location str: The location to the metadata
+
+ Returns:
+ Table: The newly registered table
+
+ Raises:
+ TableAlreadyExistsError: If the table already exists
+ """
+ raise NotImplementedError
+
def _commit_table(self, table_request: CommitTableRequest) -> CommitTableResponse:
"""Update the table.
diff --git a/python/pyiceberg/catalog/hive.py b/python/pyiceberg/catalog/hive.py
index 8942169db0..7563270d09 100644
--- a/python/pyiceberg/catalog/hive.py
+++ b/python/pyiceberg/catalog/hive.py
@@ -303,6 +303,21 @@ class HiveCatalog(Catalog):
return self._convert_hive_into_iceberg(hive_table, io)
+ def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
+ """Register a new table using existing metadata.
+
+ Args:
+ identifier Union[str, Identifier]: Table identifier for the table
+ metadata_location str: The location to the metadata
+
+ Returns:
+ Table: The newly registered table
+
+ Raises:
+ TableAlreadyExistsError: If the table already exists
+ """
+ raise NotImplementedError
+
def _commit_table(self, table_request: CommitTableRequest) -> CommitTableResponse:
"""Update the table.
diff --git a/python/pyiceberg/catalog/noop.py b/python/pyiceberg/catalog/noop.py
index bb93772aa7..083f851d1c 100644
--- a/python/pyiceberg/catalog/noop.py
+++ b/python/pyiceberg/catalog/noop.py
@@ -49,6 +49,21 @@ class NoopCatalog(Catalog):
def load_table(self, identifier: Union[str, Identifier]) -> Table:
raise NotImplementedError
+ def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
+ """Register a new table using existing metadata.
+
+ Args:
+ identifier Union[str, Identifier]: Table identifier for the table
+ metadata_location str: The location to the metadata
+
+ Returns:
+ Table: The newly registered table
+
+ Raises:
+ TableAlreadyExistsError: If the table already exists
+ """
+ raise NotImplementedError
+
def drop_table(self, identifier: Union[str, Identifier]) -> None:
raise NotImplementedError
diff --git a/python/pyiceberg/catalog/rest.py b/python/pyiceberg/catalog/rest.py
index 60c0e57484..0023e18984 100644
--- a/python/pyiceberg/catalog/rest.py
+++ b/python/pyiceberg/catalog/rest.py
@@ -79,6 +79,7 @@ class Endpoints:
update_namespace_properties: str = "namespaces/{namespace}/properties"
list_tables: str = "namespaces/{namespace}/tables"
create_table: str = "namespaces/{namespace}/tables"
+ register_table = "namespaces/{namespace}/register"
load_table: str = "namespaces/{namespace}/tables/{table}"
update_table: str = "namespaces/{namespace}/tables/{table}"
drop_table: str = "namespaces/{namespace}/tables/{table}?purgeRequested={purge}"
@@ -127,6 +128,11 @@ class CreateTableRequest(IcebergBaseModel):
properties: Properties = Field(default_factory=dict)
+class RegisterTableRequest(IcebergBaseModel):
+ name: str
+ metadata_location: str = Field(..., alias="metadata-location")
+
+
class TokenResponse(IcebergBaseModel):
access_token: str = Field()
token_type: str = Field()
@@ -443,6 +449,37 @@ class RestCatalog(Catalog):
table_response = TableResponse(**response.json())
return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
+ def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
+ """Register a new table using existing metadata.
+
+ Args:
+ identifier Union[str, Identifier]: Table identifier for the table
+ metadata_location str: The location to the metadata
+
+ Returns:
+ Table: The newly registered table
+
+ Raises:
+ TableAlreadyExistsError: If the table already exists
+ """
+ namespace_and_table = self._split_identifier_for_path(identifier)
+ request = RegisterTableRequest(
+ name=namespace_and_table["table"],
+ metadata_location=metadata_location,
+ )
+ serialized_json = request.model_dump_json().encode("utf-8")
+ response = self._session.post(
+ self.url(Endpoints.register_table, namespace=namespace_and_table["namespace"]),
+ data=serialized_json,
+ )
+ try:
+ response.raise_for_status()
+ except HTTPError as exc:
+ self._handle_non_200_response(exc, {409: TableAlreadyExistsError})
+
+ table_response = TableResponse(**response.json())
+ return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
+
def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
namespace_tuple = self._check_valid_namespace_identifier(namespace)
namespace_concat = NAMESPACE_SEPARATOR.join(namespace_tuple)
diff --git a/python/pyiceberg/catalog/sql.py b/python/pyiceberg/catalog/sql.py
index 4cc5b224a2..56e460c49e 100644
--- a/python/pyiceberg/catalog/sql.py
+++ b/python/pyiceberg/catalog/sql.py
@@ -181,6 +181,21 @@ class SqlCatalog(Catalog):
return self.load_table(identifier=identifier)
+ def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
+ """Register a new table using existing metadata.
+
+ Args:
+ identifier Union[str, Identifier]: Table identifier for the table
+ metadata_location str: The location to the metadata
+
+ Returns:
+ Table: The newly registered table
+
+ Raises:
+ TableAlreadyExistsError: If the table already exists
+ """
+ raise NotImplementedError
+
def load_table(self, identifier: Union[str, Identifier]) -> Table:
"""Load the table's metadata and return the table instance.
diff --git a/python/tests/catalog/test_base.py b/python/tests/catalog/test_base.py
index e4da808014..d1d7d638a6 100644
--- a/python/tests/catalog/test_base.py
+++ b/python/tests/catalog/test_base.py
@@ -110,6 +110,9 @@ class InMemoryCatalog(Catalog):
self.__tables[identifier] = table
return table
+ def register_table(self, identifier: Union[str, Identifier], metadata_location: str) -> Table:
+ raise NotImplementedError
+
def _commit_table(self, table_request: CommitTableRequest) -> CommitTableResponse:
new_metadata: Optional[TableMetadata] = None
metadata_location = ""
diff --git a/python/tests/catalog/test_rest.py b/python/tests/catalog/test_rest.py
index 829611e988..1c7581d24a 100644
--- a/python/tests/catalog/test_rest.py
+++ b/python/tests/catalog/test_rest.py
@@ -677,6 +677,142 @@ def test_create_table_409(rest_mock: Mocker, table_schema_simple: Schema) -> Non
assert "Table already exists" in str(e.value)
+def test_register_table_200(rest_mock: Mocker, table_schema_simple: Schema) -> None:
+ rest_mock.post(
+ f"{TEST_URI}v1/namespaces/default/register",
+ json={
+ "metadata-location": "s3://warehouse/database/table/metadata.json",
+ "metadata": {
+ "format-version": 1,
+ "table-uuid": "bf289591-dcc0-4234-ad4f-5c3eed811a29",
+ "location": "s3://warehouse/database/table",
+ "last-updated-ms": 1657810967051,
+ "last-column-id": 3,
+ "schema": {
+ "type": "struct",
+ "schema-id": 0,
+ "identifier-field-ids": [2],
+ "fields": [
+ {"id": 1, "name": "foo", "required": False, "type": "string"},
+ {"id": 2, "name": "bar", "required": True, "type": "int"},
+ {"id": 3, "name": "baz", "required": False, "type": "boolean"},
+ ],
+ },
+ "current-schema-id": 0,
+ "schemas": [
+ {
+ "type": "struct",
+ "schema-id": 0,
+ "identifier-field-ids": [2],
+ "fields": [
+ {"id": 1, "name": "foo", "required": False, "type": "string"},
+ {"id": 2, "name": "bar", "required": True, "type": "int"},
+ {"id": 3, "name": "baz", "required": False, "type": "boolean"},
+ ],
+ }
+ ],
+ "partition-spec": [],
+ "default-spec-id": 0,
+ "last-partition-id": 999,
+ "default-sort-order-id": 0,
+ "sort-orders": [{"order-id": 0, "fields": []}],
+ "properties": {
+ "write.delete.parquet.compression-codec": "zstd",
+ "write.metadata.compression-codec": "gzip",
+ "write.summary.partition-limit": "100",
+ "write.parquet.compression-codec": "zstd",
+ },
+ "current-snapshot-id": -1,
+ "refs": {},
+ "snapshots": [],
+ "snapshot-log": [],
+ "metadata-log": [],
+ },
+ "config": {
+ "client.factory": "io.tabular.iceberg.catalog.TabularAwsClientFactory",
+ "region": "us-west-2",
+ },
+ },
+ status_code=200,
+ request_headers=TEST_HEADERS,
+ )
+ catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+ actual = catalog.register_table(
+ identifier=("default", "registered_table"), metadata_location="s3://warehouse/database/table/metadata.json"
+ )
+ expected = Table(
+ identifier=("rest", "default", "registered_table"),
+ metadata_location="s3://warehouse/database/table/metadata.json",
+ metadata=TableMetadataV1(
+ location="s3://warehouse/database/table",
+ table_uuid=UUID("bf289591-dcc0-4234-ad4f-5c3eed811a29"),
+ last_updated_ms=1657810967051,
+ last_column_id=3,
+ schemas=[
+ Schema(
+ NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
+ NestedField(field_id=2, name="bar", field_type=IntegerType(), required=True),
+ NestedField(field_id=3, name="baz", field_type=BooleanType(), required=False),
+ schema_id=0,
+ identifier_field_ids=[2],
+ )
+ ],
+ current_schema_id=0,
+ default_spec_id=0,
+ last_partition_id=999,
+ properties={
+ "write.delete.parquet.compression-codec": "zstd",
+ "write.metadata.compression-codec": "gzip",
+ "write.summary.partition-limit": "100",
+ "write.parquet.compression-codec": "zstd",
+ },
+ current_snapshot_id=None,
+ snapshots=[],
+ snapshot_log=[],
+ metadata_log=[],
+ sort_orders=[SortOrder(order_id=0)],
+ default_sort_order_id=0,
+ refs={},
+ format_version=1,
+ schema_=Schema(
+ NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
+ NestedField(field_id=2, name="bar", field_type=IntegerType(), required=True),
+ NestedField(field_id=3, name="baz", field_type=BooleanType(), required=False),
+ schema_id=0,
+ identifier_field_ids=[2],
+ ),
+ partition_spec=[],
+ ),
+ io=load_file_io(),
+ catalog=catalog,
+ )
+ assert actual.metadata.model_dump() == expected.metadata.model_dump()
+ assert actual.metadata_location == expected.metadata_location
+ assert actual.identifier == expected.identifier
+
+
+def test_register_table_409(rest_mock: Mocker, table_schema_simple: Schema) -> None:
+ rest_mock.post(
+ f"{TEST_URI}v1/namespaces/default/register",
+ json={
+ "error": {
+ "message": "Table already exists: fokko.fokko2 in warehouse 8bcb0838-50fc-472d-9ddb-8feb89ef5f1e",
+ "type": "AlreadyExistsException",
+ "code": 409,
+ }
+ },
+ status_code=409,
+ request_headers=TEST_HEADERS,
+ )
+
+ catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN)
+ with pytest.raises(TableAlreadyExistsError) as e:
+ catalog.register_table(
+ identifier=("default", "registered_table"), metadata_location="s3://warehouse/database/table/metadata.json"
+ )
+ assert "Table already exists" in str(e.value)
+
+
def test_delete_namespace_204(rest_mock: Mocker) -> None:
namespace = "example"
rest_mock.delete(