Add ViewMetadata read support and create_view to REST Catalog (#2154)

<!--
Thanks for opening a pull request!
-->

<!-- In the case this PR will resolve an issue, please replace
${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
<!-- Closes #${GITHUB_ISSUE_ID} -->
Part of #818 

This adds `create_view` to the REST Catalog. As part of this, it also
defines the View and ViewMetadata class.

# Rationale for this change
PyIceberg's REST Catalog doesn't support `create_view`. Furthermore,
PyIceberg doesn't support views at all. This is the first part in
getting views supported.

# Are these changes tested?
Unit tests included

# Are there any user-facing changes?
Added Iceberg REST Catalog support for `create_view`

<!-- In the case of user-facing changes, please add the changelog label.
-->

---------

Co-authored-by: Sung Yun <107272191+sungwy@users.noreply.github.com>
Co-authored-by: Fokko Driesprong <fokko@apache.org>

Author: 3 people

pyiceberg/catalog/__init__.py

@@ -68,6 +68,8 @@
 )
 from pyiceberg.utils.config import Config, merge_config
 from pyiceberg.utils.properties import property_as_bool
+from pyiceberg.view import View
+from pyiceberg.view.metadata import ViewVersion
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -683,6 +685,31 @@ def drop_view(self, identifier: str | Identifier) -> None:
             NoSuchViewError: If a view with the given name does not exist.
         """
 
+    @abstractmethod
+    def create_view(
+        self,
+        identifier: str | Identifier,
+        schema: Schema | pa.Schema,
+        view_version: ViewVersion,
+        location: str | None = None,
+        properties: Properties = EMPTY_DICT,
+    ) -> View:
+        """Create a view.
+
+        Args:
+            identifier (str | Identifier): View identifier.
+            schema (Schema): View's schema.
+            view_version (ViewVersion): The format version for the view.
+            location (str | None): Location for the view. Optional Argument.
+            properties (Properties): View properties that can be a string based dictionary.
+
+        Returns:
+            View: the created view instance.
+
+        Raises:
+            ViewAlreadyExistsError: If a view with the name already exists.
+        """
+
     @staticmethod
     def identifier_to_tuple(identifier: str | Identifier) -> Identifier:
         """Parse an identifier to a tuple.
@@ -903,6 +930,16 @@ def purge_table(self, identifier: str | Identifier) -> None:
         delete_files(io, prev_metadata_files, PREVIOUS_METADATA)
         delete_files(io, {table.metadata_location}, METADATA)
 
+    def create_view(
+        self,
+        identifier: str | Identifier,
+        schema: Schema | pa.Schema,
+        view_version: ViewVersion,
+        location: str | None = None,
+        properties: Properties = EMPTY_DICT,
+    ) -> View:
+        raise NotImplementedError
+
     def _create_staged_table(
         self,
         identifier: str | Identifier,

pyiceberg/catalog/bigquery_metastore.py

@@ -14,8 +14,10 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from __future__ import annotations
+
 import json
-from typing import TYPE_CHECKING, Any, Union
+from typing import TYPE_CHECKING, Any
 
 from google.api_core.exceptions import NotFound
 from google.cloud.bigquery import Client, Dataset, DatasetReference, TableReference
@@ -101,7 +103,7 @@ def __init__(self, name: str, **properties: str):
     def create_table(
         self,
         identifier: str | Identifier,
-        schema: Union[Schema, "pa.Schema"],
+        schema: Schema | pa.Schema,
         location: str | None = None,
         partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
@@ -272,7 +274,7 @@ def register_table(self, identifier: str | Identifier, metadata_location: str) -
         """Register a new table using existing metadata.
 
         Args:
-            identifier (Union[str, Identifier]): Table identifier for the table
+            identifier (str | Identifier): Table identifier for the table
             metadata_location (str): The location to the metadata
 
         Returns:

pyiceberg/catalog/dynamodb.py

@@ -59,6 +59,8 @@
 )
 from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties
 from pyiceberg.utils.properties import get_first_property_value
+from pyiceberg.view import View
+from pyiceberg.view.metadata import ViewVersion
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -537,6 +539,16 @@ def update_namespace_properties(
 
         return properties_update_summary
 
+    def create_view(
+        self,
+        identifier: str | Identifier,
+        schema: Union[Schema, "pa.Schema"],
+        view_version: ViewVersion,
+        location: str | None = None,
+        properties: Properties = EMPTY_DICT,
+    ) -> View:
+        raise NotImplementedError
+
     def list_views(self, namespace: str | Identifier) -> list[Identifier]:
         raise NotImplementedError
 

pyiceberg/catalog/glue.py

@@ -85,6 +85,8 @@
     UUIDType,
 )
 from pyiceberg.utils.properties import get_first_property_value, property_as_bool
+from pyiceberg.view import View
+from pyiceberg.view.metadata import ViewVersion
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -809,6 +811,16 @@ def update_namespace_properties(
 
         return properties_update_summary
 
+    def create_view(
+        self,
+        identifier: str | Identifier,
+        schema: Union[Schema, "pa.Schema"],
+        view_version: ViewVersion,
+        location: str | None = None,
+        properties: Properties = EMPTY_DICT,
+    ) -> View:
+        raise NotImplementedError
+
     def list_views(self, namespace: str | Identifier) -> list[Identifier]:
         raise NotImplementedError
 

pyiceberg/catalog/hive.py

@@ -110,6 +110,8 @@
     UUIDType,
 )
 from pyiceberg.utils.properties import property_as_bool, property_as_float
+from pyiceberg.view import View
+from pyiceberg.view.metadata import ViewVersion
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -434,6 +436,16 @@ def create_table(
 
         return self._convert_hive_into_iceberg(hive_table)
 
+    def create_view(
+        self,
+        identifier: str | Identifier,
+        schema: Union[Schema, "pa.Schema"],
+        view_version: ViewVersion,
+        location: str | None = None,
+        properties: Properties = EMPTY_DICT,
+    ) -> View:
+        raise NotImplementedError
+
     def register_table(self, identifier: str | Identifier, metadata_location: str) -> Table:
         """Register a new table using existing metadata.
 

pyiceberg/catalog/noop.py

@@ -14,9 +14,10 @@
 #  KIND, either express or implied.  See the License for the
 #  specific language governing permissions and limitations
 #  under the License.
+from __future__ import annotations
+
 from typing import (
     TYPE_CHECKING,
-    Union,
 )
 
 from pyiceberg.catalog import Catalog, PropertiesUpdateSummary
@@ -33,6 +34,8 @@
     TableUpdate,
 )
 from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties
+from pyiceberg.view import View
+from pyiceberg.view.metadata import ViewVersion
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -42,7 +45,7 @@ class NoopCatalog(Catalog):
     def create_table(
         self,
         identifier: str | Identifier,
-        schema: Union[Schema, "pa.Schema"],
+        schema: Schema | pa.Schema,
         location: str | None = None,
         partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
@@ -53,7 +56,7 @@ def create_table(
     def create_table_transaction(
         self,
         identifier: str | Identifier,
-        schema: Union[Schema, "pa.Schema"],
+        schema: Schema | pa.Schema,
         location: str | None = None,
         partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
@@ -130,3 +133,13 @@ def namespace_exists(self, namespace: str | Identifier) -> bool:
 
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
+
+    def create_view(
+        self,
+        identifier: str | Identifier,
+        schema: Schema | pa.Schema,
+        view_version: ViewVersion,
+        location: str | None = None,
+        properties: Properties = EMPTY_DICT,
+    ) -> View:
+        raise NotImplementedError

pyiceberg/catalog/rest/__init__.py

@@ -14,12 +14,13 @@
 #  KIND, either express or implied.  See the License for the
 #  specific language governing permissions and limitations
 #  under the License.
+from __future__ import annotations
+
 from collections import deque
 from enum import Enum
 from typing import (
     TYPE_CHECKING,
     Any,
-    Union,
 )
 from urllib.parse import quote, unquote
 
@@ -55,6 +56,7 @@
     NoSuchViewError,
     TableAlreadyExistsError,
     UnauthorizedError,
+    ViewAlreadyExistsError,
 )
 from pyiceberg.io import (
     AWS_ACCESS_KEY_ID,
@@ -87,6 +89,8 @@
 from pyiceberg.types import transform_dict_value_to_str
 from pyiceberg.utils.deprecated import deprecation_message
 from pyiceberg.utils.properties import get_first_property_value, get_header_properties, property_as_bool, property_as_int
+from pyiceberg.view import View
+from pyiceberg.view.metadata import ViewMetadata, ViewVersion
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -123,7 +127,7 @@ def __str__(self) -> str:
         return f"{self.http_method.value} {self.path}"
 
     @classmethod
-    def from_string(cls, endpoint: str) -> "Endpoint":
+    def from_string(cls, endpoint: str) -> Endpoint:
         elements = endpoint.strip().split(None, 1)
         if len(elements) != 2:
             raise ValueError(f"Invalid endpoint (must consist of two elements separated by a single space): {endpoint}")
@@ -148,6 +152,7 @@ class Endpoints:
     get_token: str = "oauth/tokens"
     rename_table: str = "tables/rename"
     list_views: str = "namespaces/{namespace}/views"
+    create_view: str = "namespaces/{namespace}/views"
     drop_view: str = "namespaces/{namespace}/views/{view}"
     view_exists: str = "namespaces/{namespace}/views/{view}"
     plan_table_scan: str = "namespaces/{namespace}/tables/{table}/plan"
@@ -275,6 +280,12 @@ class TableResponse(IcebergBaseModel):
     storage_credentials: list[StorageCredential] = Field(alias="storage-credentials", default_factory=list)
 
 
+class ViewResponse(IcebergBaseModel):
+    metadata_location: str | None = Field(alias="metadata-location", default=None)
+    metadata: ViewMetadata
+    config: Properties = Field(default_factory=dict)
+
+
 class CreateTableRequest(IcebergBaseModel):
     name: str = Field()
     location: str | None = Field()
@@ -290,6 +301,18 @@ def transform_properties_dict_value_to_str(cls, properties: Properties) -> dict[
         return transform_dict_value_to_str(properties)
 
 
+class CreateViewRequest(IcebergBaseModel):
+    name: str = Field()
+    location: str | None = Field()
+    view_schema: Schema = Field(alias="schema")
+    view_version: ViewVersion = Field(alias="view-version")
+    properties: Properties = Field(default_factory=dict)
+
+    @field_validator("properties", mode="before")
+    def transform_properties_dict_value_to_str(cls, properties: Properties) -> dict[str, str]:
+        return transform_dict_value_to_str(properties)
+
+
 class RegisterTableRequest(IcebergBaseModel):
     name: str
     metadata_location: str = Field(..., alias="metadata-location")
@@ -800,6 +823,12 @@ def _response_to_staged_table(self, identifier_tuple: tuple[str, ...], table_res
             catalog=self,
         )
 
+    def _response_to_view(self, identifier_tuple: tuple[str, ...], view_response: ViewResponse) -> View:
+        return View(
+            identifier=identifier_tuple,
+            metadata=view_response.metadata,
+        )
+
     def _refresh_token(self) -> None:
         # Reactive token refresh is atypical - we should proactively refresh tokens in a separate thread
         # instead of retrying on Auth Exceptions. Keeping refresh behavior for the LegacyOAuth2AuthManager
@@ -819,7 +848,7 @@ def _config_headers(self, session: Session) -> None:
     def _create_table(
         self,
         identifier: str | Identifier,
-        schema: Union[Schema, "pa.Schema"],
+        schema: Schema | pa.Schema,
         location: str | None = None,
         partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
@@ -862,7 +891,7 @@ def _create_table(
     def create_table(
         self,
         identifier: str | Identifier,
-        schema: Union[Schema, "pa.Schema"],
+        schema: Schema | pa.Schema,
         location: str | None = None,
         partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
@@ -883,7 +912,7 @@ def create_table(
     def create_table_transaction(
         self,
         identifier: str | Identifier,
-        schema: Union[Schema, "pa.Schema"],
+        schema: Schema | pa.Schema,
         location: str | None = None,
         partition_spec: PartitionSpec = UNPARTITIONED_PARTITION_SPEC,
         sort_order: SortOrder = UNSORTED_SORT_ORDER,
@@ -901,6 +930,44 @@ def create_table_transaction(
         staged_table = self._response_to_staged_table(self.identifier_to_tuple(identifier), table_response)
         return CreateTableTransaction(staged_table)
 
+    @retry(**_RETRY_ARGS)
+    def create_view(
+        self,
+        identifier: str | Identifier,
+        schema: Schema | pa.Schema,
+        view_version: ViewVersion,
+        location: str | None = None,
+        properties: Properties = EMPTY_DICT,
+    ) -> View:
+        iceberg_schema = self._convert_schema_if_needed(schema)
+        fresh_schema = assign_fresh_schema_ids(iceberg_schema)
+
+        namespace_and_view = self._split_identifier_for_path(identifier, IdentifierKind.VIEW)
+        if location:
+            location = location.rstrip("/")
+
+        request = CreateViewRequest(
+            name=namespace_and_view["view"],
+            location=location,
+            view_schema=fresh_schema,
+            view_version=view_version,
+            properties=properties,
+        )
+
+        serialized_json = request.model_dump_json().encode(UTF8)
+        response = self._session.post(
+            self.url(Endpoints.create_view, namespace=namespace_and_view["namespace"]),
+            data=serialized_json,
+        )
+
+        try:
+            response.raise_for_status()
+        except HTTPError as exc:
+            _handle_non_200_response(exc, {409: ViewAlreadyExistsError})
+
+        view_response = ViewResponse.model_validate_json(response.text)
+        return self._response_to_view(self.identifier_to_tuple(identifier), view_response)
+
     @retry(**_RETRY_ARGS)
     def register_table(self, identifier: str | Identifier, metadata_location: str) -> Table:
         """Register a new table using existing metadata.