[#5438] feat(iceberg): add view event for Iceberg REST server (#5584)

### What changes were proposed in this pull request?

Add view event for Iceberg REST server

### Why are the changes needed?

Close: #5438

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

unit test

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

@@ -31,6 +31,9 @@
 import org.apache.gravitino.iceberg.service.dispatcher.IcebergTableEventDispatcher;
 import org.apache.gravitino.iceberg.service.dispatcher.IcebergTableOperationDispatcher;
 import org.apache.gravitino.iceberg.service.dispatcher.IcebergTableOperationExecutor;
+import org.apache.gravitino.iceberg.service.dispatcher.IcebergViewEventDispatcher;
+import org.apache.gravitino.iceberg.service.dispatcher.IcebergViewOperationDispatcher;
+import org.apache.gravitino.iceberg.service.dispatcher.IcebergViewOperationExecutor;
 import org.apache.gravitino.iceberg.service.metrics.IcebergMetricsManager;
 import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider;
 import org.apache.gravitino.iceberg.service.provider.IcebergConfigProviderFactory;
@@ -90,6 +93,10 @@ private void initServer(IcebergConfig icebergConfig) {
         new IcebergTableOperationExecutor(icebergCatalogWrapperManager);
     IcebergTableEventDispatcher icebergTableEventDispatcher =
         new IcebergTableEventDispatcher(icebergTableOperationExecutor, eventBus, metalakeName);
+    IcebergViewOperationExecutor icebergViewOperationExecutor =
+        new IcebergViewOperationExecutor(icebergCatalogWrapperManager);
+    IcebergViewEventDispatcher icebergViewEventDispatcher =
+        new IcebergViewEventDispatcher(icebergViewOperationExecutor, eventBus, metalakeName);
 
     config.register(
         new AbstractBinder() {
@@ -98,6 +105,7 @@ protected void configure() {
             bind(icebergCatalogWrapperManager).to(IcebergCatalogWrapperManager.class).ranked(1);
             bind(icebergMetricsManager).to(IcebergMetricsManager.class).ranked(1);
             bind(icebergTableEventDispatcher).to(IcebergTableOperationDispatcher.class).ranked(1);
+            bind(icebergViewEventDispatcher).to(IcebergViewOperationDispatcher.class).ranked(1);
           }
         });
 

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

@@ -0,0 +1,210 @@
+/*
+ *  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.dispatcher;
+
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.iceberg.service.IcebergRestUtils;
+import org.apache.gravitino.listener.EventBus;
+import org.apache.gravitino.listener.api.event.IcebergCreateViewEvent;
+import org.apache.gravitino.listener.api.event.IcebergCreateViewFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergCreateViewPreEvent;
+import org.apache.gravitino.listener.api.event.IcebergDropViewEvent;
+import org.apache.gravitino.listener.api.event.IcebergDropViewFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergDropViewPreEvent;
+import org.apache.gravitino.listener.api.event.IcebergListViewEvent;
+import org.apache.gravitino.listener.api.event.IcebergListViewFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergListViewPreEvent;
+import org.apache.gravitino.listener.api.event.IcebergLoadViewEvent;
+import org.apache.gravitino.listener.api.event.IcebergLoadViewFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergLoadViewPreEvent;
+import org.apache.gravitino.listener.api.event.IcebergRenameViewEvent;
+import org.apache.gravitino.listener.api.event.IcebergRenameViewFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergRenameViewPreEvent;
+import org.apache.gravitino.listener.api.event.IcebergReplaceViewEvent;
+import org.apache.gravitino.listener.api.event.IcebergReplaceViewFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergReplaceViewPreEvent;
+import org.apache.gravitino.listener.api.event.IcebergRequestContext;
+import org.apache.gravitino.listener.api.event.IcebergViewExistsEvent;
+import org.apache.gravitino.listener.api.event.IcebergViewExistsFailureEvent;
+import org.apache.gravitino.listener.api.event.IcebergViewExistsPreEvent;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.CreateViewRequest;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.ListTablesResponse;
+import org.apache.iceberg.rest.responses.LoadViewResponse;
+
+/**
+ * {@code IcebergViewEventDispatcher} is a decorator for {@link IcebergViewOperationExecutor} that
+ * not only delegates view operations to the underlying dispatcher but also dispatches corresponding
+ * events to an {@link EventBus}.
+ */
+public class IcebergViewEventDispatcher implements IcebergViewOperationDispatcher {
+
+  private IcebergViewOperationDispatcher icebergViewOperationDispatcher;
+  private EventBus eventBus;
+  private String metalakeName;
+
+  public IcebergViewEventDispatcher(
+      IcebergViewOperationDispatcher icebergViewOperationDispatcher,
+      EventBus eventBus,
+      String metalakeName) {
+    this.icebergViewOperationDispatcher = icebergViewOperationDispatcher;
+    this.eventBus = eventBus;
+    this.metalakeName = metalakeName;
+  }
+
+  @Override
+  public LoadViewResponse createView(
+      IcebergRequestContext context, Namespace namespace, CreateViewRequest createViewRequest) {
+    TableIdentifier viewIdentifier = TableIdentifier.of(namespace, createViewRequest.name());
+    NameIdentifier nameIdentifier =
+        IcebergRestUtils.getGravitinoNameIdentifier(
+            metalakeName, context.catalogName(), viewIdentifier);
+    eventBus.dispatchEvent(
+        new IcebergCreateViewPreEvent(context, nameIdentifier, createViewRequest));
+    LoadViewResponse loadViewResponse;
+    try {
+      loadViewResponse =
+          icebergViewOperationDispatcher.createView(context, namespace, createViewRequest);
+    } catch (Exception e) {
+      eventBus.dispatchEvent(
+          new IcebergCreateViewFailureEvent(context, nameIdentifier, createViewRequest, e));
+      throw e;
+    }
+    eventBus.dispatchEvent(
+        new IcebergCreateViewEvent(context, nameIdentifier, createViewRequest, loadViewResponse));
+    return loadViewResponse;
+  }
+
+  @Override
+  public LoadViewResponse replaceView(
+      IcebergRequestContext context,
+      TableIdentifier viewIdentifier,
+      UpdateTableRequest replaceViewRequest) {
+    NameIdentifier gravitinoNameIdentifier =
+        IcebergRestUtils.getGravitinoNameIdentifier(
+            metalakeName, context.catalogName(), viewIdentifier);
+    eventBus.dispatchEvent(
+        new IcebergReplaceViewPreEvent(context, gravitinoNameIdentifier, replaceViewRequest));
+    LoadViewResponse loadViewResponse;
+    try {
+      loadViewResponse =
+          icebergViewOperationDispatcher.replaceView(context, viewIdentifier, replaceViewRequest);
+    } catch (Exception e) {
+      eventBus.dispatchEvent(
+          new IcebergReplaceViewFailureEvent(
+              context, gravitinoNameIdentifier, replaceViewRequest, e));
+      throw e;
+    }
+    eventBus.dispatchEvent(
+        new IcebergReplaceViewEvent(
+            context, gravitinoNameIdentifier, replaceViewRequest, loadViewResponse));
+    return loadViewResponse;
+  }
+
+  @Override
+  public void dropView(IcebergRequestContext context, TableIdentifier viewIdentifier) {
+    NameIdentifier gravitinoNameIdentifier =
+        IcebergRestUtils.getGravitinoNameIdentifier(
+            metalakeName, context.catalogName(), viewIdentifier);
+    eventBus.dispatchEvent(new IcebergDropViewPreEvent(context, gravitinoNameIdentifier));
+    try {
+      icebergViewOperationDispatcher.dropView(context, viewIdentifier);
+    } catch (Exception e) {
+      eventBus.dispatchEvent(new IcebergDropViewFailureEvent(context, gravitinoNameIdentifier, e));
+      throw e;
+    }
+    eventBus.dispatchEvent(new IcebergDropViewEvent(context, gravitinoNameIdentifier));
+  }
+
+  @Override
+  public LoadViewResponse loadView(IcebergRequestContext context, TableIdentifier viewIdentifier) {
+    NameIdentifier gravitinoNameIdentifier =
+        IcebergRestUtils.getGravitinoNameIdentifier(
+            metalakeName, context.catalogName(), viewIdentifier);
+    eventBus.dispatchEvent(new IcebergLoadViewPreEvent(context, gravitinoNameIdentifier));
+    LoadViewResponse loadViewResponse;
+    try {
+      loadViewResponse = icebergViewOperationDispatcher.loadView(context, viewIdentifier);
+    } catch (Exception e) {
+      eventBus.dispatchEvent(new IcebergLoadViewFailureEvent(context, gravitinoNameIdentifier, e));
+      throw e;
+    }
+    eventBus.dispatchEvent(
+        new IcebergLoadViewEvent(context, gravitinoNameIdentifier, loadViewResponse));
+    return loadViewResponse;
+  }
+
+  @Override
+  public ListTablesResponse listView(IcebergRequestContext context, Namespace namespace) {
+    NameIdentifier gravitinoNameIdentifier =
+        IcebergRestUtils.getGravitinoNameIdentifier(metalakeName, context.catalogName(), namespace);
+    eventBus.dispatchEvent(new IcebergListViewPreEvent(context, gravitinoNameIdentifier));
+    ListTablesResponse listViewsResponse;
+    try {
+      listViewsResponse = icebergViewOperationDispatcher.listView(context, namespace);
+    } catch (Exception e) {
+      eventBus.dispatchEvent(new IcebergListViewFailureEvent(context, gravitinoNameIdentifier, e));
+      throw e;
+    }
+    eventBus.dispatchEvent(new IcebergListViewEvent(context, gravitinoNameIdentifier));
+    return listViewsResponse;
+  }
+
+  @Override
+  public boolean viewExists(IcebergRequestContext context, TableIdentifier viewIdentifier) {
+    NameIdentifier gravitinoNameIdentifier =
+        IcebergRestUtils.getGravitinoNameIdentifier(
+            metalakeName, context.catalogName(), viewIdentifier);
+    eventBus.dispatchEvent(new IcebergViewExistsPreEvent(context, gravitinoNameIdentifier));
+    boolean isExists;
+    try {
+      isExists = icebergViewOperationDispatcher.viewExists(context, viewIdentifier);
+    } catch (Exception e) {
+      eventBus.dispatchEvent(
+          new IcebergViewExistsFailureEvent(context, gravitinoNameIdentifier, e));
+      throw e;
+    }
+    eventBus.dispatchEvent(new IcebergViewExistsEvent(context, gravitinoNameIdentifier, isExists));
+    return isExists;
+  }
+
+  @Override
+  public void renameView(IcebergRequestContext context, RenameTableRequest renameViewRequest) {
+    TableIdentifier sourceView = renameViewRequest.source();
+    NameIdentifier gravitinoNameIdentifier =
+        IcebergRestUtils.getGravitinoNameIdentifier(
+            metalakeName, context.catalogName(), sourceView);
+    eventBus.dispatchEvent(
+        new IcebergRenameViewPreEvent(context, gravitinoNameIdentifier, renameViewRequest));
+    try {
+      icebergViewOperationDispatcher.renameView(context, renameViewRequest);
+    } catch (Exception e) {
+      eventBus.dispatchEvent(
+          new IcebergRenameViewFailureEvent(
+              context, gravitinoNameIdentifier, renameViewRequest, e));
+      throw e;
+    }
+    eventBus.dispatchEvent(
+        new IcebergRenameViewEvent(context, gravitinoNameIdentifier, renameViewRequest));
+  }
+}

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

@@ -0,0 +1,103 @@
+/*
+ *  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.dispatcher;
+
+import org.apache.gravitino.listener.api.event.IcebergRequestContext;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.rest.requests.CreateViewRequest;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.apache.iceberg.rest.responses.ListTablesResponse;
+import org.apache.iceberg.rest.responses.LoadViewResponse;
+
+/**
+ * The {@code IcebergViewOperationDispatcher} interface defines the public API for managing Iceberg
+ * views.
+ */
+public interface IcebergViewOperationDispatcher {
+
+  /**
+   * Creates a new Iceberg view.
+   *
+   * @param context Iceberg REST request context information.
+   * @param namespace The namespace within which the view should be created.
+   * @param createViewRequest The request object containing the details for creating the view.
+   * @return A {@link LoadViewResponse} object containing the result of the operation.
+   */
+  LoadViewResponse createView(
+      IcebergRequestContext context, Namespace namespace, CreateViewRequest createViewRequest);
+
+  /**
+   * Updates an Iceberg view.
+   *
+   * @param context Iceberg REST request context information.
+   * @param viewIdentifier The Iceberg view identifier.
+   * @param replaceViewRequest The request object containing the details for updating the view.
+   * @return A {@link LoadViewResponse} object containing the result of the operation.
+   */
+  LoadViewResponse replaceView(
+      IcebergRequestContext context,
+      TableIdentifier viewIdentifier,
+      UpdateTableRequest replaceViewRequest);
+
+  /**
+   * Drops an Iceberg view.
+   *
+   * @param context Iceberg REST request context information.
+   * @param viewIdentifier The Iceberg view identifier.
+   */
+  void dropView(IcebergRequestContext context, TableIdentifier viewIdentifier);
+
+  /**
+   * Loads an Iceberg view.
+   *
+   * @param context Iceberg REST request context information.
+   * @param viewIdentifier The Iceberg view identifier.
+   * @return A {@link LoadViewResponse} object containing the result of the operation.
+   */
+  LoadViewResponse loadView(IcebergRequestContext context, TableIdentifier viewIdentifier);
+
+  /**
+   * Lists Iceberg views.
+   *
+   * @param context Iceberg REST request context information.
+   * @param namespace The Iceberg namespace.
+   * @return A {@link ListTablesResponse} object containing the list of view identifiers.
+   */
+  ListTablesResponse listView(IcebergRequestContext context, Namespace namespace);
+
+  /**
+   * Check whether an Iceberg view exists.
+   *
+   * @param context Iceberg REST request context information.
+   * @param viewIdentifier The Iceberg view identifier.
+   * @return Whether view exists.
+   */
+  boolean viewExists(IcebergRequestContext context, TableIdentifier viewIdentifier);
+
+  /**
+   * Rename an Iceberg view.
+   *
+   * @param context Iceberg REST request context information.
+   * @param renameViewRequest Rename view request information.
+   */
+  void renameView(IcebergRequestContext context, RenameTableRequest renameViewRequest);
+}