[#4370]feat(iceberg): support view interface for Iceberg REST server

iceberg/iceberg-common/src/main/java/org/apache/gravitino/iceberg/common/ops/IcebergCatalogWrapper.java

@@ -43,9 +43,11 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.catalog.ViewCatalog;
 import org.apache.iceberg.rest.CatalogHandlers;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.CreateViewRequest;
 import org.apache.iceberg.rest.requests.RegisterTableRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
@@ -55,6 +57,7 @@
 import org.apache.iceberg.rest.responses.ListNamespacesResponse;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
+import org.apache.iceberg.rest.responses.LoadViewResponse;
 import org.apache.iceberg.rest.responses.UpdateNamespacePropertiesResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -116,6 +119,13 @@ private void validateNamespace(Optional<Namespace> namespace) {
     }
   }
 
+  private ViewCatalog getViewCatalog() {
+    if (!(catalog instanceof ViewCatalog)) {
+      throw new UnsupportedOperationException(catalog.name() + " is not support view");
+    }
+    return (ViewCatalog) catalog;
+  }
+
   public CreateNamespaceResponse createNamespace(CreateNamespaceRequest request) {
     validateNamespace(Optional.of(request.namespace()));
     return CatalogHandlers.createNamespace(asNamespaceCatalog, request);
@@ -203,6 +213,37 @@ public LoadTableResponse updateTable(IcebergTableChange icebergTableChange) {
     return loadTable(icebergTableChange.getTableIdentifier());
   }
 
+  public LoadViewResponse createView(Namespace namespace, CreateViewRequest request) {
+    request.validate();
+    return CatalogHandlers.createView(getViewCatalog(), namespace, request);
+  }
+
+  public LoadViewResponse updateView(TableIdentifier viewIdentifier, UpdateTableRequest request) {
+    request.validate();
+    return CatalogHandlers.updateView(getViewCatalog(), viewIdentifier, request);
+  }
+
+  public LoadViewResponse loadView(TableIdentifier viewIdentifier) {
+    return CatalogHandlers.loadView(getViewCatalog(), viewIdentifier);
+  }
+
+  public void dropView(TableIdentifier viewIdentifier) {
+    CatalogHandlers.dropView(getViewCatalog(), viewIdentifier);
+  }
+
+  public void renameView(RenameTableRequest request) {
+    request.validate();
+    CatalogHandlers.renameView(getViewCatalog(), request);
+  }
+
+  public boolean existView(TableIdentifier viewIdentifier) {
+    return getViewCatalog().viewExists(viewIdentifier);
+  }
+
+  public ListTablesResponse listView(Namespace namespace) {
+    return CatalogHandlers.listViews(getViewCatalog(), namespace);
+  }
+
   @Override
   public void close() throws Exception {
     if (catalog instanceof AutoCloseable) {

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/IcebergExceptionMapper.java

@@ -32,6 +32,7 @@
 import org.apache.iceberg.exceptions.NoSuchIcebergTableException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.NoSuchViewException;
 import org.apache.iceberg.exceptions.NotAuthorizedException;
 import org.apache.iceberg.exceptions.ServiceUnavailableException;
 import org.apache.iceberg.exceptions.UnprocessableEntityException;
@@ -57,6 +58,7 @@ public class IcebergExceptionMapper implements ExceptionMapper<Exception> {
           .put(NoSuchTableException.class, 404)
           .put(NoSuchIcebergTableException.class, 404)
           .put(UnsupportedOperationException.class, 406)
+          .put(NoSuchViewException.class, 404)
           .put(AlreadyExistsException.class, 409)
           .put(CommitFailedException.class, 409)
           .put(UnprocessableEntityException.class, 422)

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergViewOperations.java

@@ -0,0 +1,149 @@
+/*
+ * 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.
+ */
+package org.apache.gravitino.iceberg.service.rest;
+
+import com.codahale.metrics.annotation.ResponseMetered;
+import com.codahale.metrics.annotation.Timed;
+import javax.inject.Inject;
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.GET;
+import javax.ws.rs.HEAD;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
+import org.apache.gravitino.iceberg.service.IcebergRestUtils;
+import org.apache.gravitino.metrics.MetricNames;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.RESTUtil;
+import org.apache.iceberg.rest.requests.CreateViewRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.ListTablesResponse;
+import org.apache.iceberg.rest.responses.LoadViewResponse;
+
+@Path("/v1/{prefix:([^/]*/)?}namespaces/{namespace}/views")
+@Consumes(MediaType.APPLICATION_JSON)
+@Produces(MediaType.APPLICATION_JSON)
+public class IcebergViewOperations {
+
+  private IcebergCatalogWrapperManager icebergCatalogWrapperManager;
+
+  @SuppressWarnings("UnusedVariable")
+  @Context
+  private HttpServletRequest httpRequest;
+
+  @Inject
+  public IcebergViewOperations(IcebergCatalogWrapperManager icebergCatalogWrapperManager) {
+    this.icebergCatalogWrapperManager = icebergCatalogWrapperManager;
+  }
+
+  @GET
+  @Produces(MediaType.APPLICATION_JSON)
+  @Timed(name = "list-view." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
+  @ResponseMetered(name = "list-view", absolute = true)
+  public Response listView(
+      @PathParam("prefix") String prefix, @PathParam("namespace") String namespace) {
+    ListTablesResponse response =
+        icebergCatalogWrapperManager.getOps(prefix).listView(RESTUtil.decodeNamespace(namespace));
+    return IcebergRestUtils.ok(response);
+  }
+
+  @POST
+  @Produces(MediaType.APPLICATION_JSON)
+  @Timed(name = "create-view." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
+  @ResponseMetered(name = "create-view", absolute = true)
+  public Response createView(
+      @PathParam("prefix") String prefix,
+      @PathParam("namespace") String namespace,
+      CreateViewRequest request) {
+    LoadViewResponse response =
+        icebergCatalogWrapperManager
+            .getOps(prefix)
+            .createView(RESTUtil.decodeNamespace(namespace), request);
+    return IcebergRestUtils.ok(response);
+  }
+
+  @GET
+  @Path("{view}")
+  @Produces(MediaType.APPLICATION_JSON)
+  @Timed(name = "load-view." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
+  @ResponseMetered(name = "load-view", absolute = true)
+  public Response loadView(
+      @PathParam("prefix") String prefix,
+      @PathParam("namespace") String namespace,
+      @PathParam("view") String view) {
+    TableIdentifier viewIdentifier = TableIdentifier.of(RESTUtil.decodeNamespace(namespace), view);
+    LoadViewResponse response =
+        icebergCatalogWrapperManager.getOps(prefix).loadView(viewIdentifier);
+    return IcebergRestUtils.ok(response);
+  }
+
+  @POST
+  @Path("{view}")
+  @Produces(MediaType.APPLICATION_JSON)
+  @Timed(name = "replace-view." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
+  @ResponseMetered(name = "replace-view", absolute = true)
+  public Response replaceView(
+      @PathParam("prefix") String prefix,
+      @PathParam("namespace") String namespace,
+      @PathParam("view") String view,
+      UpdateTableRequest request) {
+    TableIdentifier viewIdentifier = TableIdentifier.of(RESTUtil.decodeNamespace(namespace), view);
+    LoadViewResponse response =
+        icebergCatalogWrapperManager.getOps(prefix).updateView(viewIdentifier, request);
+    return IcebergRestUtils.ok(response);
+  }
+
+  @DELETE
+  @Path("{view}")
+  @Produces(MediaType.APPLICATION_JSON)
+  @Timed(name = "drop-view." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
+  @ResponseMetered(name = "drop-view", absolute = true)
+  public Response dropView(
+      @PathParam("prefix") String prefix,
+      @PathParam("namespace") String namespace,
+      @PathParam("view") String view) {
+    TableIdentifier viewIdentifier = TableIdentifier.of(RESTUtil.decodeNamespace(namespace), view);
+    icebergCatalogWrapperManager.getOps(prefix).dropView(viewIdentifier);
+    return IcebergRestUtils.noContent();
+  }
+
+  @HEAD
+  @Path("{view}")
+  @Produces(MediaType.APPLICATION_JSON)
+  @Timed(name = "view-exists." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
+  @ResponseMetered(name = "view-exits", absolute = true)
+  public Response viewExists(
+      @PathParam("prefix") String prefix,
+      @PathParam("namespace") String namespace,
+      @PathParam("view") String view) {
+    TableIdentifier tableIdentifier = TableIdentifier.of(RESTUtil.decodeNamespace(namespace), view);
+    if (icebergCatalogWrapperManager.getOps(prefix).existView(tableIdentifier)) {
+      return IcebergRestUtils.noContent();
+    } else {
+      return IcebergRestUtils.notExists();
+    }
+  }
+}

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergViewRenameOperations.java

@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+package org.apache.gravitino.iceberg.service.rest;
+
+import com.codahale.metrics.annotation.ResponseMetered;
+import com.codahale.metrics.annotation.Timed;
+import javax.inject.Inject;
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager;
+import org.apache.gravitino.iceberg.service.IcebergRestUtils;
+import org.apache.gravitino.metrics.MetricNames;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+
+@Path("/v1/{prefix:([^/]*/)?}views/rename")
+@Consumes(MediaType.APPLICATION_JSON)
+@Produces(MediaType.APPLICATION_JSON)
+public class IcebergViewRenameOperations {
+
+  @SuppressWarnings("UnusedVariable")
+  @Context
+  private HttpServletRequest httpRequest;
+
+  private IcebergCatalogWrapperManager icebergCatalogWrapperManager;
+
+  @Inject
+  public IcebergViewRenameOperations(IcebergCatalogWrapperManager icebergCatalogWrapperManager) {
+    this.icebergCatalogWrapperManager = icebergCatalogWrapperManager;
+  }
+
+  @POST
+  @Produces(MediaType.APPLICATION_JSON)
+  @Timed(name = "rename-view." + MetricNames.HTTP_PROCESS_DURATION, absolute = true)
+  @ResponseMetered(name = "rename-view", absolute = true)
+  public Response renameView(@PathParam("prefix") String prefix, RenameTableRequest request) {
+    icebergCatalogWrapperManager.getOps(prefix).renameView(request);
+    return IcebergRestUtils.noContent();
+  }
+}

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergRestTestUtil.java

@@ -44,7 +44,11 @@ public class IcebergRestTestUtil {
   public static final String UPDATE_NAMESPACE_POSTFIX = "properties";
   public static final String TEST_NAMESPACE_NAME = "gravitino-test";
   public static final String TABLE_PATH = NAMESPACE_PATH + "/" + TEST_NAMESPACE_NAME + "/tables";
+
+  public static final String VIEW_PATH = NAMESPACE_PATH + "/" + TEST_NAMESPACE_NAME + "/views";
   public static final String RENAME_TABLE_PATH = V_1 + "/tables/rename";
+
+  public static final String RENAME_VIEW_PATH = V_1 + "/views/rename";
   public static final String REPORT_METRICS_POSTFIX = "metrics";
 
   public static final boolean DEBUG_SERVER_LOG_ENABLED = true;

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergTestBase.java

@@ -45,16 +45,30 @@ public Invocation.Builder getRenameTableClientBuilder() {
     return getIcebergClientBuilder(IcebergRestTestUtil.RENAME_TABLE_PATH, Optional.empty());
   }
 
+  public Invocation.Builder getRenameViewClientBuilder() {
+    return getIcebergClientBuilder(IcebergRestTestUtil.RENAME_VIEW_PATH, Optional.empty());
+  }
+
   public Invocation.Builder getTableClientBuilder() {
     return getTableClientBuilder(Optional.empty());
   }
 
+  public Invocation.Builder getViewClientBuilder() {
+    return getViewClientBuilder(Optional.empty());
+  }
+
   public Invocation.Builder getTableClientBuilder(Optional<String> name) {
     String path =
         Joiner.on("/").skipNulls().join(IcebergRestTestUtil.TABLE_PATH, name.orElseGet(() -> null));
     return getIcebergClientBuilder(path, Optional.empty());
   }
 
+  public Invocation.Builder getViewClientBuilder(Optional<String> name) {
+    String path =
+        Joiner.on("/").skipNulls().join(IcebergRestTestUtil.VIEW_PATH, name.orElseGet(() -> null));
+    return getIcebergClientBuilder(path, Optional.empty());
+  }
+
   public Invocation.Builder getReportMetricsClientBuilder(String name) {
     String path =
         Joiner.on("/")