This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 941f030f3e [#12116] improvement(client-python): Add view create/drop
operations (#12117)
941f030f3e is described below
commit 941f030f3e0f624351fda392f39fd969c081c893
Author: Zhiguo Wu <[email protected]>
AuthorDate: Wed Jul 22 12:53:39 2026 +0800
[#12116] improvement(client-python): Add view create/drop operations
(#12117)
### What changes were proposed in this pull request?
- Extend `RelationalCatalog` with `ViewCatalog`.
- Add `as_view_catalog`, `create_view`, and `drop_view`.
- Add request and response DTOs for view creation.
- Add view-specific REST error handling.
- Generalize the entity namespace helper for table and view operations.
- Add corresponding unit and integration tests.
### Why are the changes needed?
The Python client currently lacks support for view create/drop
operations. These changes allow users to create and drop views through
the Python client.
Fix: #12116
### Does this PR introduce _any_ user-facing change?
Yes. Python client users can now create and drop views through
`RelationalCatalog.as_view_catalog()`.
### How was this patch tested?
Added unit tests for request and response validation, REST error
handling, and view create/drop operations.
Added integration tests covering view creation, duplicate creation, view
deletion, and deletion of nonexistent views.
---
.../gravitino/client/relational_catalog.py | 151 ++++++++++++++++++---
.../gravitino/dto/requests/view_create_request.py | 96 +++++++++++++
.../gravitino/dto/responses/view_response.py | 49 +++++++
.../exceptions/handlers/view_error_handler.py | 78 +++++++++++
.../tests/integration/test_relational_catalog.py | 86 ++++++++++++
.../dto/requests/test_view_create_request.py | 144 ++++++++++++++++++++
.../unittests/dto/responses/test_view_response.py | 51 +++++++
.../tests/unittests/test_error_handler.py | 83 +++++++++++
.../tests/unittests/test_relational_catalog.py | 93 +++++++++++++
9 files changed, 811 insertions(+), 20 deletions(-)
diff --git a/clients/client-python/gravitino/client/relational_catalog.py
b/clients/client-python/gravitino/client/relational_catalog.py
index 40fbc9914c..26562e1e9a 100644
--- a/clients/client-python/gravitino/client/relational_catalog.py
+++ b/clients/client-python/gravitino/client/relational_catalog.py
@@ -24,19 +24,28 @@ from
gravitino.api.rel.expressions.distributions.distribution import Distributio
from gravitino.api.rel.expressions.sorts.sort_order import SortOrder
from gravitino.api.rel.expressions.transforms.transform import Transform
from gravitino.api.rel.indexes.index import Index
+from gravitino.api.rel.representation import Representation
from gravitino.api.rel.table import Table
from gravitino.api.rel.table_catalog import TableCatalog
+from gravitino.api.rel.view import View
+from gravitino.api.rel.view_catalog import ViewCatalog
+from gravitino.api.rel.view_change import ViewChange
from gravitino.client.base_schema_catalog import BaseSchemaCatalog
+from gravitino.client.generic_view import GenericView
from gravitino.client.relational_table import RelationalTable
from gravitino.dto.audit_dto import AuditDTO
from gravitino.dto.rel.distribution_dto import DistributionDTO
from gravitino.dto.requests.table_create_request import TableCreateRequest
from gravitino.dto.requests.table_updates_request import TableUpdatesRequest
+from gravitino.dto.requests.view_create_request import ViewCreateRequest
from gravitino.dto.responses.drop_response import DropResponse
from gravitino.dto.responses.entity_list_response import EntityListResponse
from gravitino.dto.responses.table_response import TableResponse
+from gravitino.dto.responses.view_response import ViewResponse
from gravitino.dto.util.dto_converters import DTOConverters
+from gravitino.exceptions.base import UnsupportedOperationException
from gravitino.exceptions.handlers.table_error_handler import
TABLE_ERROR_HANDLER
+from gravitino.exceptions.handlers.view_error_handler import VIEW_ERROR_HANDLER
from gravitino.name_identifier import NameIdentifier
from gravitino.namespace import Namespace
from gravitino.rest.rest_utils import encode_string
@@ -44,13 +53,13 @@ from gravitino.utils import HTTPClient
class RelationalCatalog(
- BaseSchemaCatalog, TableCatalog
+ BaseSchemaCatalog, TableCatalog, ViewCatalog
): # pylint: disable=too-many-ancestors
"""Relational catalog is a catalog implementation
- The `RelationalCatalog` supports relational database like metadata
operations,
- for example, schemas and tables list, creation, update and deletion. A
Relational
- catalog is under the metalake.
+ The `RelationalCatalog` supports relational metadata operations such as
listing,
+ creating, updating, and deleting schemas, tables, and views. A relational
catalog
+ belongs to a metalake.
"""
PRIVILEGES: Final[str] = "privileges"
@@ -88,6 +97,32 @@ class RelationalCatalog(
"""
return self
+ def as_view_catalog(self) -> ViewCatalog:
+ """Return this relational catalog as a :class:`ViewCatalog`.
+
+ This method returns ``self`` to provide access to view-related
+ operations defined by the :class:`ViewCatalog` interface.
+
+ Returns:
+ ViewCatalog: The current catalog instance as a ``ViewCatalog``.
+ """
+ return self
+
+ def _get_entity_full_namespace(self, entity_namespace: Namespace) ->
Namespace:
+ """Get the full namespace of an entity with the given short namespace.
+
+ Args:
+ entity_namespace (Namespace): The entity's short namespace, which
is the schema name.
+
+ Returns:
+ Namespace: full namespace of the entity, which is
"metalake.catalog.schema" format.
+ """
+ return Namespace.of(
+ self._catalog_namespace.level(0),
+ self._name,
+ entity_namespace.level(0),
+ )
+
def _check_table_name_identifier(self, identifier: NameIdentifier) -> None:
"""Check whether the `NameIdentifier` of a table is valid.
@@ -119,27 +154,51 @@ class RelationalCatalog(
f"Table namespace must be non-null and have 1 level, the input
namespace is {namespace}",
)
- def _get_table_full_namespace(self, table_namespace: Namespace) ->
Namespace:
- """Get the full namespace of the table with the given table's short
namespace (schema name).
+ def _format_table_request_path(self, ns: Namespace) -> str:
+ schema_ns = Namespace.of(ns.level(0), ns.level(1))
+ return (
+ f"{BaseSchemaCatalog.format_schema_request_path(schema_ns)}"
+ f"/{encode_string(ns.level(2))}"
+ "/tables"
+ )
+
+ def _check_view_name_identifier(self, identifier: NameIdentifier) -> None:
+ """Check whether the `NameIdentifier` of a view is valid.
Args:
- table_namespace (Namespace): The table's short namespace, which is
the schema name.
+ identifier (NameIdentifier):
+ The NameIdentifier to check, which should be "schema.view"
format.
- Returns:
- Namespace: full namespace of the table, which is
"metalake.catalog.schema" format.
+ Raises:
+ IllegalNameIdentifierException: If the Namespace is not valid.
"""
- return Namespace.of(
- self._catalog_namespace.level(0),
- self._name,
- table_namespace.level(0),
+ NameIdentifier.check(identifier is not None, "NameIdentifier must not
be null")
+ NameIdentifier.check(
+ identifier.name() is not None and identifier.name() != "",
+ "NameIdentifier name must not be empty",
)
+ self._check_view_namespace(identifier.namespace())
- def _format_table_request_path(self, ns: Namespace) -> str:
+ def _check_view_namespace(self, namespace: Namespace) -> None:
+ """Check whether the namespace of a view is valid, which should be
"schema".
+
+ Args:
+ namespace (Namespace): The namespace to check.
+
+ Raises:
+ IllegalNamespaceException: If the Namespace is not valid.
+ """
+ Namespace.check(
+ namespace is not None and namespace.length() == 1,
+ f"View namespace must be non-null and have 1 level, the input
namespace is {namespace}",
+ )
+
+ def _format_view_request_path(self, ns: Namespace) -> str:
schema_ns = Namespace.of(ns.level(0), ns.level(1))
return (
f"{BaseSchemaCatalog.format_schema_request_path(schema_ns)}"
f"/{encode_string(ns.level(2))}"
- "/tables"
+ "/views"
)
def create_table(
@@ -169,7 +228,7 @@ class RelationalCatalog(
_indexes=DTOConverters.to_dtos(indexes),
)
req.validate()
- full_namespace = self._get_table_full_namespace(identifier.namespace())
+ full_namespace =
self._get_entity_full_namespace(identifier.namespace())
resp = self.rest_client.post(
self._format_table_request_path(full_namespace),
json=req,
@@ -181,7 +240,7 @@ class RelationalCatalog(
def list_tables(self, namespace: Namespace) -> list[NameIdentifier]:
self._check_table_namespace(namespace)
- full_namespace = self._get_table_full_namespace(namespace)
+ full_namespace = self._get_entity_full_namespace(namespace)
resp = self.rest_client.get(
self._format_table_request_path(full_namespace),
error_handler=TABLE_ERROR_HANDLER,
@@ -207,7 +266,7 @@ class RelationalCatalog(
required_privilege_names: Optional[set[Privilege.Name]] = None,
) -> Table:
self._check_table_name_identifier(identifier)
- full_namespace = self._get_table_full_namespace(identifier.namespace())
+ full_namespace =
self._get_entity_full_namespace(identifier.namespace())
query_params = (
{
RelationalCatalog.PRIVILEGES: ",".join(
@@ -244,7 +303,7 @@ class RelationalCatalog(
def alter_table(self, identifier: NameIdentifier, *changes) -> Table:
self._check_table_name_identifier(identifier)
- full_namespace = self._get_table_full_namespace(identifier.namespace())
+ full_namespace =
self._get_entity_full_namespace(identifier.namespace())
updates_request = TableUpdatesRequest(
updates=[
DTOConverters.to_table_update_request(change) for change in
changes
@@ -276,7 +335,7 @@ class RelationalCatalog(
def _drop_table(self, identifier: NameIdentifier, purge: bool) -> bool:
self._check_table_name_identifier(identifier)
- full_namespace = self._get_table_full_namespace(identifier.namespace())
+ full_namespace =
self._get_entity_full_namespace(identifier.namespace())
resp = self.rest_client.delete(
f"{self._format_table_request_path(full_namespace)}"
f"/{encode_string(identifier.name())}",
@@ -286,3 +345,55 @@ class RelationalCatalog(
drop_resp = DropResponse.from_json(resp.body, infer_missing=True)
drop_resp.validate()
return drop_resp.dropped()
+
+ def list_views(self, namespace: Namespace) -> list[NameIdentifier]:
+ raise UnsupportedOperationException("Listing views is not supported")
+
+ def load_view(self, identifier: NameIdentifier) -> View:
+ raise UnsupportedOperationException("Loading views is not supported")
+
+ def create_view(
+ self,
+ identifier: NameIdentifier,
+ columns: list[Column],
+ representations: list[Representation],
+ comment: Optional[str] = None,
+ default_catalog: Optional[str] = None,
+ default_schema: Optional[str] = None,
+ properties: Optional[dict[str, str]] = None,
+ ) -> View:
+ self._check_view_name_identifier(identifier)
+ req = ViewCreateRequest(
+ _name=identifier.name(),
+ _columns=DTOConverters.to_dtos(columns),
+ _representations=DTOConverters.to_dtos(representations),
+ _comment=comment,
+ _default_catalog=default_catalog,
+ _default_schema=default_schema,
+ _properties=properties,
+ )
+ req.validate()
+ full_namespace =
self._get_entity_full_namespace(identifier.namespace())
+ resp = self.rest_client.post(
+ self._format_view_request_path(full_namespace),
+ json=req,
+ error_handler=VIEW_ERROR_HANDLER,
+ )
+ view_resp = ViewResponse.from_json(resp.body, infer_missing=True)
+ view_resp.validate()
+ return GenericView(view_resp.view())
+
+ def alter_view(self, identifier: NameIdentifier, *changes: ViewChange) ->
View:
+ raise UnsupportedOperationException("View alteration is not supported")
+
+ def drop_view(self, identifier: NameIdentifier) -> bool:
+ self._check_view_name_identifier(identifier)
+ full_namespace =
self._get_entity_full_namespace(identifier.namespace())
+ resp = self.rest_client.delete(
+ f"{self._format_view_request_path(full_namespace)}"
+ f"/{encode_string(identifier.name())}",
+ error_handler=VIEW_ERROR_HANDLER,
+ )
+ drop_resp = DropResponse.from_json(resp.body, infer_missing=True)
+ drop_resp.validate()
+ return drop_resp.dropped()
diff --git
a/clients/client-python/gravitino/dto/requests/view_create_request.py
b/clients/client-python/gravitino/dto/requests/view_create_request.py
new file mode 100644
index 0000000000..700ac5b9a6
--- /dev/null
+++ b/clients/client-python/gravitino/dto/requests/view_create_request.py
@@ -0,0 +1,96 @@
+# 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.
+
+from dataclasses import dataclass, field
+from typing import Optional
+
+from dataclasses_json import config
+
+from gravitino.dto.rel.column_dto import ColumnDTO
+from gravitino.dto.rel.json_serdes.representation_serdes import
RepresentationSerdes
+from gravitino.dto.rel.representation_dto import RepresentationDTO
+from gravitino.dto.rel.sql_representation_dto import SQLRepresentationDTO
+from gravitino.rest.rest_message import RESTRequest
+from gravitino.utils.precondition import Precondition
+
+
+@dataclass
+class ViewCreateRequest(RESTRequest):
+ """Represents a request to create a view."""
+
+ _name: str = field(metadata=config(field_name="name"))
+ _columns: Optional[list[ColumnDTO]] = field(
+ default=None, metadata=config(field_name="columns")
+ )
+ _representations: Optional[list[RepresentationDTO]] = field(
+ default=None,
+ metadata=config(
+ field_name="representations",
+ encoder=lambda items: [
+ RepresentationSerdes.serialize(item) for item in items
+ ],
+ decoder=lambda values: [
+ RepresentationSerdes.deserialize(value) for value in values
+ ],
+ exclude=lambda value: value is None,
+ ),
+ )
+ _comment: Optional[str] = field(default=None,
metadata=config(field_name="comment"))
+ _default_catalog: Optional[str] = field(
+ default=None, metadata=config(field_name="defaultCatalog")
+ )
+ _default_schema: Optional[str] = field(
+ default=None, metadata=config(field_name="defaultSchema")
+ )
+ _properties: Optional[dict[str, str]] = field(
+ default=None, metadata=config(field_name="properties")
+ )
+
+ def validate(self) -> None:
+ Precondition.check_string_not_empty(
+ self._name, '"name" field is required and cannot be empty'
+ )
+ Precondition.check_argument(
+ self._representations is not None and len(self._representations) >
0,
+ '"representations" field is required and cannot be empty',
+ )
+ for representation in self._representations:
+ Precondition.check_argument(
+ representation is not None, "representation must not be null"
+ )
+ representation.validate()
+ if self._columns:
+ for column in self._columns:
+ Precondition.check_argument(
+ column is not None, "column must not be null"
+ )
+ column.validate()
+ ViewCreateRequest.validate_no_duplicate_dialects(self._representations)
+
+ @staticmethod
+ def validate_no_duplicate_dialects(
+ representations: list[RepresentationDTO],
+ ) -> None:
+ """Validate that SQL representations do not use duplicate dialects."""
+ seen_dialects = set()
+ for representation in representations:
+ if isinstance(representation, SQLRepresentationDTO):
+ Precondition.check_argument(
+ representation.dialect() not in seen_dialects,
+ f"Duplicate SQL representation dialect:
{representation.dialect()}",
+ )
+ seen_dialects.add(representation.dialect())
diff --git a/clients/client-python/gravitino/dto/responses/view_response.py
b/clients/client-python/gravitino/dto/responses/view_response.py
new file mode 100644
index 0000000000..63d2d2ca22
--- /dev/null
+++ b/clients/client-python/gravitino/dto/responses/view_response.py
@@ -0,0 +1,49 @@
+# 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.
+
+from dataclasses import dataclass, field
+
+from dataclasses_json import config
+
+from gravitino.dto.rel.view_dto import ViewDTO
+from gravitino.dto.responses.base_response import BaseResponse
+from gravitino.exceptions.base import IllegalArgumentException
+
+
+@dataclass
+class ViewResponse(BaseResponse):
+ """Response object for view-related operations."""
+
+ _view: ViewDTO = field(metadata=config(field_name="view"))
+
+ def view(self) -> ViewDTO:
+ """Returns the view DTO."""
+ return self._view
+
+ def validate(self):
+ """Validates the response data."""
+ super().validate()
+ if self._view is None:
+ raise IllegalArgumentException("view must not be null")
+ if not self._view.name():
+ raise IllegalArgumentException("view 'name' must not be null or
empty")
+ if self._view.audit_info() is None:
+ raise IllegalArgumentException("view 'audit' must not be null")
+ if not self._view.representations():
+ raise IllegalArgumentException(
+ "view 'representations' must not be null or empty"
+ )
diff --git
a/clients/client-python/gravitino/exceptions/handlers/view_error_handler.py
b/clients/client-python/gravitino/exceptions/handlers/view_error_handler.py
new file mode 100644
index 0000000000..674438c144
--- /dev/null
+++ b/clients/client-python/gravitino/exceptions/handlers/view_error_handler.py
@@ -0,0 +1,78 @@
+# 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.
+
+from gravitino.constants.error import ErrorConstants
+from gravitino.dto.responses.error_response import ErrorResponse
+from gravitino.exceptions.base import (
+ CatalogNotInUseException,
+ ForbiddenException,
+ IllegalArgumentException,
+ MetalakeNotInUseException,
+ NoSuchCatalogException,
+ NoSuchSchemaException,
+ NoSuchViewException,
+ NotFoundException,
+ NotInUseException,
+ UnsupportedOperationException,
+ ViewAlreadyExistsException,
+)
+from gravitino.exceptions.handlers.rest_error_handler import RestErrorHandler
+
+
+class ViewErrorHandler(RestErrorHandler):
+ """Error handler for view operations."""
+
+ def handle(self, error_response: ErrorResponse):
+ error_message = error_response.format_error_message()
+ code = ErrorConstants(error_response.code())
+ exception_type = error_response.type()
+
+ if code is ErrorConstants.ILLEGAL_ARGUMENTS_CODE:
+ raise IllegalArgumentException(error_message)
+
+ if code is ErrorConstants.NOT_FOUND_CODE:
+ if exception_type == NoSuchCatalogException.__name__:
+ raise NoSuchCatalogException(error_message)
+ if exception_type == NoSuchSchemaException.__name__:
+ raise NoSuchSchemaException(error_message)
+ if exception_type == NoSuchViewException.__name__:
+ raise NoSuchViewException(error_message)
+ raise NotFoundException(error_message)
+
+ if code is ErrorConstants.ALREADY_EXISTS_CODE:
+ raise ViewAlreadyExistsException(error_message)
+
+ if code is ErrorConstants.INTERNAL_ERROR_CODE:
+ raise RuntimeError(error_message)
+
+ if code is ErrorConstants.UNSUPPORTED_OPERATION_CODE:
+ raise UnsupportedOperationException(error_message)
+
+ if code is ErrorConstants.FORBIDDEN_CODE:
+ raise ForbiddenException(error_message)
+
+ if code is ErrorConstants.NOT_IN_USE_CODE:
+ if exception_type == CatalogNotInUseException.__name__:
+ raise CatalogNotInUseException(error_message)
+ if exception_type == MetalakeNotInUseException.__name__:
+ raise MetalakeNotInUseException(error_message)
+ raise NotInUseException(error_message)
+
+ super().handle(error_response)
+
+
+VIEW_ERROR_HANDLER = ViewErrorHandler()
diff --git a/clients/client-python/tests/integration/test_relational_catalog.py
b/clients/client-python/tests/integration/test_relational_catalog.py
index 0dd75eb838..1723025514 100644
--- a/clients/client-python/tests/integration/test_relational_catalog.py
+++ b/clients/client-python/tests/integration/test_relational_catalog.py
@@ -17,6 +17,7 @@
import logging
from random import randint
+from uuid import uuid4
from gravitino import (
Catalog,
@@ -25,17 +26,21 @@ from gravitino import (
NameIdentifier,
)
from gravitino.api.rel.column import Column
+from gravitino.api.rel.dialects import Dialects
from gravitino.api.rel.expressions.distributions.distributions import
Distributions
from gravitino.api.rel.expressions.transforms.transforms import Transforms
from gravitino.api.rel.indexes.indexes import Indexes
+from gravitino.api.rel.sql_representation import SQLRepresentation
from gravitino.api.rel.table import Table
from gravitino.api.rel.table_change import TableChange
from gravitino.api.rel.types.types import Types
+from gravitino.api.rel.view import View
from gravitino.client.relational_table import RelationalTable
from gravitino.exceptions.base import (
NoSuchSchemaException,
NoSuchTableException,
TableAlreadyExistsException,
+ ViewAlreadyExistsException,
)
from gravitino.namespace import Namespace
from tests.integration.containers.hdfs_container import HDFSContainer
@@ -53,6 +58,8 @@ class TestRelationalCatalog(IntegrationTestEnv):
TABLE_IDENT: NameIdentifier = NameIdentifier.of(SCHEMA_NAME, TABLE_NAME)
TABLE_COMMENT: str = "Test table for relational catalog"
TABLE_PROPERTIES = {"property1": "value1", "property2": "value2"}
+ VIEW_COMMENT: str = "Test view for relational catalog"
+ VIEW_PROPERTIES = {"view_property1": "value1", "view_property2": "value2"}
@classmethod
def setUpClass(cls):
@@ -94,6 +101,10 @@ class TestRelationalCatalog(IntegrationTestEnv):
super().tearDownClass()
def setUp(self):
+ self.view_name = f"test_view_{uuid4().hex}"
+ self.view_ident = NameIdentifier.of(
+ TestRelationalCatalog.SCHEMA_NAME, self.view_name
+ )
# Create schema for each test
TestRelationalCatalog.schema = (
TestRelationalCatalog.catalog.as_schemas().create_schema(
@@ -137,6 +148,28 @@ class TestRelationalCatalog(IntegrationTestEnv):
indexes=Indexes.EMPTY_INDEXES,
)
+ def _create_test_view(self) -> View:
+ """Create a test view with basic columns."""
+
+ view_catalog = self.catalog.as_view_catalog()
+ columns = [
+ Column.of("id", Types.LongType.get(), "Primary key"),
+ Column.of("name", Types.StringType.get(), "Name column"),
+ ]
+
+ return view_catalog.create_view(
+ identifier=self.view_ident,
+ columns=columns,
+ representations=[
+ SQLRepresentation(
+ Dialects.HIVE,
+ f"SELECT id, name FROM {TestRelationalCatalog.TABLE_NAME}",
+ )
+ ],
+ comment=TestRelationalCatalog.VIEW_COMMENT,
+ properties=TestRelationalCatalog.VIEW_PROPERTIES,
+ )
+
def test_relational_catalog_create_table(self):
"""Test creating a table in the relational catalog."""
table = self._create_test_table()
@@ -311,3 +344,56 @@ class TestRelationalCatalog(IntegrationTestEnv):
new_property_value,
)
self.assertNotIn("property2", altered_table.properties())
+
+ def test_relational_catalog_create_view(self):
+ """Test creating a view in the relational catalog."""
+ self._create_test_table()
+ view = self._create_test_view()
+ self.assertIsNotNone(view)
+ self.assertEqual(view.name(), self.view_name)
+ self.assertEqual(view.comment(), TestRelationalCatalog.VIEW_COMMENT)
+ self.assertEqual(view.properties().get("view_property1"), "value1")
+ self.assertEqual(view.properties().get("view_property2"), "value2")
+ self.assertEqual(len(view.columns()), 2)
+ self.assertEqual(view.columns()[0].name(), "id")
+ self.assertEqual(view.columns()[0].data_type(), Types.LongType.get())
+ self.assertEqual(view.columns()[1].name(), "name")
+ self.assertEqual(view.columns()[1].data_type(), Types.StringType.get())
+ self.assertEqual(
+ view.sql_for(Dialects.HIVE).sql(),
+ f"SELECT id, name FROM {TestRelationalCatalog.TABLE_NAME}",
+ )
+
+ def test_relational_catalog_create_view_already_exists(self):
+ """Test creating a view that already exists should raise exception."""
+ self._create_test_table()
+ self._create_test_view()
+ view_catalog = self.catalog.as_view_catalog()
+
+ with self.assertRaises(ViewAlreadyExistsException):
+ view_catalog.create_view(
+ identifier=self.view_ident,
+ columns=[Column.of("id", Types.LongType.get(), "Primary key")],
+ representations=[
+ SQLRepresentation(
+ Dialects.HIVE,
+ f"SELECT id FROM {TestRelationalCatalog.TABLE_NAME}",
+ )
+ ],
+ )
+
+ def test_relational_catalog_drop_view(self):
+ """Test dropping a view from the relational catalog."""
+ self._create_test_table()
+ self._create_test_view()
+ view_catalog = self.catalog.as_view_catalog()
+
+ is_dropped = view_catalog.drop_view(self.view_ident)
+ self.assertTrue(is_dropped)
+
+ def test_relational_catalog_drop_view_not_exists(self):
+ """Test dropping a view that doesn't exist should return False."""
+ view_catalog = self.catalog.as_view_catalog()
+
+ is_dropped = view_catalog.drop_view(self.view_ident)
+ self.assertFalse(is_dropped)
diff --git
a/clients/client-python/tests/unittests/dto/requests/test_view_create_request.py
b/clients/client-python/tests/unittests/dto/requests/test_view_create_request.py
new file mode 100644
index 0000000000..358dc6c195
--- /dev/null
+++
b/clients/client-python/tests/unittests/dto/requests/test_view_create_request.py
@@ -0,0 +1,144 @@
+# 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 unittest
+
+from gravitino.api.rel.dialects import Dialects
+from gravitino.api.rel.representation import Representation
+from gravitino.api.rel.types.types import Types
+from gravitino.dto.rel.column_dto import ColumnDTO
+from gravitino.dto.rel.sql_representation_dto import SQLRepresentationDTO
+from gravitino.dto.requests.view_create_request import ViewCreateRequest
+
+
+class TestViewCreateRequest(unittest.TestCase):
+ @staticmethod
+ def _request(
+ name: str = "test_view",
+ columns: list[ColumnDTO] = None,
+ representations: list[SQLRepresentationDTO] = None,
+ ) -> ViewCreateRequest:
+ return ViewCreateRequest(
+ _name=name,
+ _columns=(
+ columns
+ if columns is not None
+ else [
+ ColumnDTO(
+ _name="id",
+ _data_type=Types.IntegerType.get(),
+ _nullable=False,
+ )
+ ]
+ ),
+ _representations=(
+ representations
+ if representations is not None
+ else [
+ SQLRepresentationDTO(
+ _dialect=Dialects.TRINO,
+ _sql="SELECT id FROM test_table",
+ )
+ ]
+ ),
+ _comment="comment",
+ _default_catalog="catalog",
+ _default_schema="schema",
+ _properties={"k1": "v1"},
+ )
+
+ def test_validate_and_serialize(self):
+ request = self._request()
+
+ request.validate()
+ json_str = request.to_json()
+ deserialized = ViewCreateRequest.from_json(json_str)
+ deserialized.validate()
+
+ self.assertEqual(request, deserialized)
+ self.assertIn('"name": "test_view"', json_str)
+ self.assertIn('"columns"', json_str)
+ self.assertIn('"representations"', json_str)
+ self.assertIn(f'"type": "{Representation.TYPE_SQL}"', json_str)
+ self.assertIn(f'"dialect": "{Dialects.TRINO}"', json_str)
+ self.assertIn('"sql": "SELECT id FROM test_table"', json_str)
+ self.assertIn('"defaultCatalog": "catalog"', json_str)
+ self.assertIn('"defaultSchema": "schema"', json_str)
+
+ def test_validate_rejects_invalid_name(self):
+ request = self._request(name="")
+
+ with self.assertRaises(ValueError):
+ request.validate()
+
+ def test_validate_rejects_empty_representations(self):
+ request = self._request(representations=[])
+
+ with self.assertRaises(ValueError):
+ request.validate()
+
+ def test_serialize_excludes_missing_representations(self):
+ request = ViewCreateRequest(_name="test_view")
+
+ self.assertNotIn('"representations"', request.to_json())
+
+ def test_validate_rejects_missing_representations(self):
+ request = ViewCreateRequest(_name="test_view")
+
+ with self.assertRaisesRegex(
+ ValueError, '"representations" field is required and cannot be
empty'
+ ):
+ request.validate()
+
+ def test_validate_rejects_null_representation(self):
+ request = self._request(representations=[None])
+
+ with self.assertRaises(ValueError):
+ request.validate()
+
+ def test_validate_rejects_invalid_representation(self):
+ request = self._request(
+ representations=[SQLRepresentationDTO(_dialect="", _sql="SELECT
1")]
+ )
+
+ with self.assertRaises(ValueError):
+ request.validate()
+
+ def test_validate_rejects_null_column(self):
+ request = self._request(columns=[None])
+
+ with self.assertRaises(ValueError):
+ request.validate()
+
+ def test_validate_rejects_invalid_column(self):
+ request = self._request(
+ columns=[ColumnDTO(_name="", _data_type=Types.IntegerType.get())]
+ )
+
+ with self.assertRaises(ValueError):
+ request.validate()
+
+ def test_validate_rejects_duplicate_dialect(self):
+ request = self._request(
+ representations=[
+ SQLRepresentationDTO(_dialect=Dialects.TRINO, _sql="SELECT 1"),
+ SQLRepresentationDTO(_dialect=Dialects.TRINO, _sql="SELECT 2"),
+ ]
+ )
+
+ with self.assertRaises(ValueError):
+ request.validate()
diff --git
a/clients/client-python/tests/unittests/dto/responses/test_view_response.py
b/clients/client-python/tests/unittests/dto/responses/test_view_response.py
new file mode 100644
index 0000000000..210383c57b
--- /dev/null
+++ b/clients/client-python/tests/unittests/dto/responses/test_view_response.py
@@ -0,0 +1,51 @@
+# 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 unittest
+
+from gravitino.api.rel.dialects import Dialects
+from gravitino.dto.audit_dto import AuditDTO
+from gravitino.dto.rel.sql_representation_dto import SQLRepresentationDTO
+from gravitino.dto.rel.view_dto import ViewDTO
+from gravitino.dto.responses.view_response import ViewResponse
+from gravitino.exceptions.base import IllegalArgumentException
+
+
+class TestViewResponse(unittest.TestCase):
+ @staticmethod
+ def _view_dto() -> ViewDTO:
+ return ViewDTO(
+ _name="test_view",
+ _representations=[
+ SQLRepresentationDTO(_dialect=Dialects.TRINO, _sql="SELECT 1")
+ ],
+ _audit=AuditDTO("creator"),
+ )
+
+ def test_view_response(self):
+ response = ViewResponse(_code=0, _view=self._view_dto())
+
+ response.validate()
+ deserialized = ViewResponse.from_json(response.to_json())
+
+ self.assertEqual("test_view", deserialized.view().name())
+ self.assertEqual("creator", deserialized.view().audit_info().creator())
+ self.assertEqual(1, len(deserialized.view().representations()))
+
+ def test_view_response_validate(self):
+ with self.assertRaises(IllegalArgumentException):
+ ViewResponse(_code=0, _view=None).validate()
diff --git a/clients/client-python/tests/unittests/test_error_handler.py
b/clients/client-python/tests/unittests/test_error_handler.py
index a7e52b443f..64c190e43d 100644
--- a/clients/client-python/tests/unittests/test_error_handler.py
+++ b/clients/client-python/tests/unittests/test_error_handler.py
@@ -43,6 +43,7 @@ from gravitino.exceptions.base import (
NoSuchSchemaException,
NoSuchTableException,
NoSuchUserException,
+ NoSuchViewException,
NotEmptyException,
NotFoundException,
NotInUseException,
@@ -53,6 +54,7 @@ from gravitino.exceptions.base import (
TableAlreadyExistsException,
UnsupportedOperationException,
UserAlreadyExistsException,
+ ViewAlreadyExistsException,
GroupAlreadyExistsException,
)
from gravitino.exceptions.handlers.catalog_error_handler import
CATALOG_ERROR_HANDLER
@@ -73,6 +75,7 @@ from gravitino.exceptions.handlers.role_error_handler import
ROLE_ERROR_HANDLER
from gravitino.exceptions.handlers.schema_error_handler import
SCHEMA_ERROR_HANDLER
from gravitino.exceptions.handlers.table_error_handler import
TABLE_ERROR_HANDLER
from gravitino.exceptions.handlers.user_error_handler import USER_ERROR_HANDLER
+from gravitino.exceptions.handlers.view_error_handler import VIEW_ERROR_HANDLER
class TestErrorHandler(unittest.TestCase):
@@ -435,6 +438,86 @@ class TestErrorHandler(unittest.TestCase):
ErrorResponse.generate_error_response(Exception, "mock error")
)
+ def test_view_error_handler(self):
+ with self.assertRaises(IllegalArgumentException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ IllegalArgumentException, "mock error"
+ )
+ )
+
+ with self.assertRaises(NoSuchCatalogException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ NoSuchCatalogException, "mock error"
+ )
+ )
+
+ with self.assertRaises(NoSuchSchemaException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ NoSuchSchemaException, "mock error"
+ )
+ )
+
+ with self.assertRaises(NoSuchViewException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(NoSuchViewException,
"mock error")
+ )
+
+ with self.assertRaises(NotFoundException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(NotFoundException, "mock
error")
+ )
+
+ with self.assertRaises(ViewAlreadyExistsException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ ViewAlreadyExistsException, "mock error"
+ )
+ )
+
+ with self.assertRaises(RuntimeError):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(RuntimeError, "mock
error")
+ )
+
+ with self.assertRaises(UnsupportedOperationException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ UnsupportedOperationException, "mock error"
+ )
+ )
+
+ with self.assertRaises(ForbiddenException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(ForbiddenException,
"mock error")
+ )
+
+ with self.assertRaises(CatalogNotInUseException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ CatalogNotInUseException, "mock error"
+ )
+ )
+
+ with self.assertRaises(MetalakeNotInUseException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(
+ MetalakeNotInUseException, "mock error"
+ )
+ )
+
+ with self.assertRaises(NotInUseException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(NotInUseException, "mock
error")
+ )
+
+ with self.assertRaises(RESTException):
+ VIEW_ERROR_HANDLER.handle(
+ ErrorResponse.generate_error_response(Exception, "mock error")
+ )
+
def test_user_error_handler(self):
with self.assertRaises(NoSuchMetalakeException):
USER_ERROR_HANDLER.handle(
diff --git a/clients/client-python/tests/unittests/test_relational_catalog.py
b/clients/client-python/tests/unittests/test_relational_catalog.py
index 4a9fbecc71..316f5f83df 100644
--- a/clients/client-python/tests/unittests/test_relational_catalog.py
+++ b/clients/client-python/tests/unittests/test_relational_catalog.py
@@ -20,19 +20,28 @@ from http.client import HTTPResponse
from unittest.mock import Mock, patch
from gravitino.api.authorization.privileges import Privilege
+from gravitino.api.rel.column import Column
+from gravitino.api.rel.dialects import Dialects
+from gravitino.api.rel.sql_representation import SQLRepresentation
from gravitino.api.rel.table_change import TableChange
+from gravitino.api.rel.types.types import Types
from gravitino.client.relational_catalog import RelationalCatalog
from gravitino.dto.audit_dto import AuditDTO
+from gravitino.dto.rel.column_dto import ColumnDTO
from gravitino.dto.rel.distribution_dto import DistributionDTO
+from gravitino.dto.rel.sql_representation_dto import SQLRepresentationDTO
from gravitino.dto.rel.table_dto import TableDTO
+from gravitino.dto.rel.view_dto import ViewDTO
from gravitino.dto.responses.drop_response import DropResponse
from gravitino.dto.responses.entity_list_response import EntityListResponse
from gravitino.dto.responses.table_response import TableResponse
+from gravitino.dto.responses.view_response import ViewResponse
from gravitino.dto.util.dto_converters import DTOConverters
from gravitino.exceptions.base import (
NoSuchSchemaException,
NoSuchTableException,
TableAlreadyExistsException,
+ ViewAlreadyExistsException,
)
from gravitino.name_identifier import NameIdentifier
from gravitino.namespace import Namespace
@@ -51,6 +60,8 @@ class TestRelationalCatalog(unittest.TestCase):
cls.table_name = "test_table"
cls.catalog_namespace = Namespace.of(cls.metalake_name)
cls.table_identifier = NameIdentifier.of(cls.schema_name,
cls.table_name)
+ cls.view_name = "test_view"
+ cls.view_identifier = NameIdentifier.of(cls.schema_name, cls.view_name)
cls.rest_client = HTTPClient("http://localhost:8090";)
cls.catalog = RelationalCatalog(
catalog_namespace=cls.catalog_namespace,
@@ -62,6 +73,30 @@ class TestRelationalCatalog(unittest.TestCase):
)
cls.TABLE_DTO_JSON_STRING =
TABLE_DTO_JSON_STRING_WITH_STARTING_DATE_SORT
cls.table_dto = TableDTO.from_json(cls.TABLE_DTO_JSON_STRING)
+ cls.view_dto = ViewDTO(
+ _name=cls.view_name,
+ _columns=[
+ ColumnDTO(
+ _name="id",
+ _data_type=Types.IntegerType.get(),
+ _comment="id column",
+ _nullable=False,
+ )
+ ],
+ _representations=[
+ SQLRepresentationDTO(
+ _dialect=Dialects.TRINO,
+ _sql="SELECT id FROM test_table",
+ )
+ ],
+ _comment="test view comment",
+ _default_catalog="test_catalog",
+ _default_schema="test_schema",
+ _properties={"k1": "v1"},
+ _audit=AuditDTO(
+ "creator", "2022-01-01T00:00:00Z", "modifier",
"2022-01-01T00:00:00Z"
+ ),
+ )
def _get_mock_http_resp(self, json_str: str, return_code: int = 200):
mock_http_resp = Mock(HTTPResponse)
@@ -76,6 +111,10 @@ class TestRelationalCatalog(unittest.TestCase):
table_catalog = self.catalog.as_table_catalog()
self.assertIs(table_catalog, self.catalog)
+ def test_as_view_catalog(self):
+ view_catalog = self.catalog.as_view_catalog()
+ self.assertIs(view_catalog, self.catalog)
+
def test_create_table(self):
resp_body = TableResponse(0, self.table_dto)
mock_resp = self._get_mock_http_resp(resp_body.to_json())
@@ -329,3 +368,57 @@ class TestRelationalCatalog(unittest.TestCase):
TableChange.update_column_nullability(["id"], nullable=True),
)
self.assertEqual(table.name(), self.table_dto.name())
+
+ def test_create_view(self):
+ resp_body = ViewResponse(0, self.view_dto)
+ mock_resp = self._get_mock_http_resp(resp_body.to_json())
+
+ with patch(
+ "gravitino.utils.http_client.HTTPClient.post",
+ return_value=mock_resp,
+ ):
+ view = self.catalog.as_view_catalog().create_view(
+ self.view_identifier,
+ [Column.of("id", Types.IntegerType.get(), nullable=False)],
+ [SQLRepresentation(Dialects.TRINO, "SELECT id FROM
test_table")],
+ comment="test view comment",
+ default_catalog="test_catalog",
+ default_schema="test_schema",
+ properties={"k1": "v1"},
+ )
+ self.assertEqual(self.view_name, view.name())
+ self.assertEqual("test view comment", view.comment())
+
+ def test_create_view_already_exists(self):
+ with patch(
+ "gravitino.utils.http_client.HTTPClient.post",
+ side_effect=ViewAlreadyExistsException("View already exists"),
+ ):
+ with self.assertRaises(ViewAlreadyExistsException):
+ self.catalog.as_view_catalog().create_view(
+ self.view_identifier,
+ [Column.of("id", Types.IntegerType.get())],
+ [SQLRepresentation(Dialects.TRINO, "SELECT id FROM
test_table")],
+ )
+
+ def test_drop_view(self):
+ resp_body = DropResponse(0, True)
+ mock_resp = self._get_mock_http_resp(resp_body.to_json())
+
+ with patch(
+ "gravitino.utils.http_client.HTTPClient.delete",
+ return_value=mock_resp,
+ ):
+ is_dropped =
self.catalog.as_view_catalog().drop_view(self.view_identifier)
+ self.assertTrue(is_dropped)
+
+ def test_drop_view_not_exists(self):
+ resp_body = DropResponse(0, False)
+ mock_resp = self._get_mock_http_resp(resp_body.to_json())
+
+ with patch(
+ "gravitino.utils.http_client.HTTPClient.delete",
+ return_value=mock_resp,
+ ):
+ is_dropped =
self.catalog.as_view_catalog().drop_view(self.view_identifier)
+ self.assertFalse(is_dropped)
