Core: Add View support for REST catalog

api/src/main/java/org/apache/iceberg/catalog/ViewCatalog.java

@@ -92,6 +92,7 @@ default boolean viewExists(TableIdentifier identifier) {
    * @param to new view identifier
    * @throws NoSuchViewException if the "from" view does not exist
    * @throws AlreadyExistsException if the "to" view already exists
+   * @throws NoSuchNamespaceException if the "to" namespace doesn't exist
    */
   void renameView(TableIdentifier from, TableIdentifier to);
 

api/src/main/java/org/apache/iceberg/catalog/ViewSessionCatalog.java

@@ -0,0 +1,120 @@
+/*
+ * 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.iceberg.catalog;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.NoSuchNamespaceException;
+import org.apache.iceberg.exceptions.NoSuchViewException;
+import org.apache.iceberg.view.View;
+import org.apache.iceberg.view.ViewBuilder;
+
+/** A session Catalog API for view create, drop, and load operations. */
+public interface ViewSessionCatalog {
+
+  /**
+   * Return the name for this catalog.
+   *
+   * @return this catalog's name
+   */
+  String name();
+
+  /**
+   * Return all the identifiers under this namespace.
+   *
+   * @param namespace a namespace
+   * @return a list of identifiers for views
+   * @throws NoSuchNamespaceException if the namespace is not found
+   */
+  List<TableIdentifier> listViews(SessionCatalog.SessionContext context, Namespace namespace);
+
+  /**
+   * Load a view.
+   *
+   * @param identifier a view identifier
+   * @return instance of {@link View} implementation referred by the identifier
+   * @throws NoSuchViewException if the view does not exist
+   */
+  View loadView(SessionCatalog.SessionContext context, TableIdentifier identifier);
+
+  /**
+   * Check whether view exists.
+   *
+   * @param identifier a view identifier
+   * @return true if the view exists, false otherwise
+   */
+  default boolean viewExists(SessionCatalog.SessionContext context, TableIdentifier identifier) {
+    try {
+      loadView(context, identifier);
+      return true;
+    } catch (NoSuchViewException e) {
+      return false;
+    }
+  }
+
+  /**
+   * Instantiate a builder to create or replace a SQL view.
+   *
+   * @param identifier a view identifier
+   * @return a view builder
+   */
+  ViewBuilder buildView(SessionCatalog.SessionContext context, TableIdentifier identifier);
+
+  /**
+   * Drop a view.
+   *
+   * @param identifier a view identifier
+   * @return true if the view was dropped, false if the view did not exist
+   */
+  boolean dropView(SessionCatalog.SessionContext context, TableIdentifier identifier);
+
+  /**
+   * Rename a view.
+   *
+   * @param from identifier of the view to rename
+   * @param to new view identifier
+   * @throws NoSuchViewException if the "from" view does not exist
+   * @throws AlreadyExistsException if the "to" view already exists
+   * @throws NoSuchNamespaceException if the "to" namespace doesn't exist
+   */
+  void renameView(SessionCatalog.SessionContext context, TableIdentifier from, TableIdentifier to);
+
+  /**
+   * Invalidate cached view metadata from current catalog.
+   *
+   * <p>If the view is already loaded or cached, drop cached data. If the view does not exist or is
+   * not cached, do nothing.
+   *
+   * @param identifier a view identifier
+   */
+  default void invalidateView(SessionCatalog.SessionContext context, TableIdentifier identifier) {}
+
+  /**
+   * Initialize a view catalog given a custom name and a map of catalog properties.
+   *
+   * <p>A custom view catalog implementation must have a no-arg constructor. A compute engine like
+   * Spark or Flink will first initialize the catalog without any arguments, and then call this
+   * method to complete catalog initialization with properties passed into the engine.
+   *
+   * @param name a custom name for the catalog
+   * @param properties catalog properties
+   */
+  void initialize(String name, Map<String, String> properties);
+}

core/src/main/java/org/apache/iceberg/UpdateRequirement.java

@@ -19,12 +19,19 @@
 package org.apache.iceberg;
 
 import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.view.ViewMetadata;
 
 /** Represents a requirement for a {@link MetadataUpdate} */
 public interface UpdateRequirement {
   void validate(TableMetadata base);
 
+  default void validate(ViewMetadata base) {
+    throw new ValidationException(
+        "Cannot validate %s against a view", this.getClass().getSimpleName());
+  }
+
   class AssertTableDoesNotExist implements UpdateRequirement {
     public AssertTableDoesNotExist() {}
 
@@ -55,6 +62,14 @@ public void validate(TableMetadata base) {
             "Requirement failed: UUID does not match: expected %s != %s", base.uuid(), uuid);
       }
     }
+
+    @Override
+    public void validate(ViewMetadata base) {
+      if (!uuid.equalsIgnoreCase(base.uuid())) {
+        throw new CommitFailedException(
+            "Requirement failed: UUID does not match: expected %s != %s", base.uuid(), uuid);
+      }
+    }
   }
 
   class AssertRefSnapshotID implements UpdateRequirement {

core/src/main/java/org/apache/iceberg/catalog/BaseViewSessionCatalog.java

@@ -0,0 +1,92 @@
+/*
+ * 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.iceberg.catalog;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.iceberg.view.View;
+import org.apache.iceberg.view.ViewBuilder;
+
+public abstract class BaseViewSessionCatalog extends BaseSessionCatalog
+    implements ViewSessionCatalog {
+
+  private final Cache<String, ViewCatalog> catalogs =
+      Caffeine.newBuilder().expireAfterAccess(10, TimeUnit.MINUTES).build();
+
+  public ViewCatalog asViewCatalog(SessionContext context) {
+    return catalogs.get(context.sessionId(), id -> new AsViewCatalog(context));
+  }
+
+  public class AsViewCatalog implements ViewCatalog {
+    private final SessionContext context;
+
+    private AsViewCatalog(SessionContext context) {
+      this.context = context;
+    }
+
+    @Override
+    public String name() {
+      return BaseViewSessionCatalog.this.name();
+    }
+
+    @Override
+    public List<TableIdentifier> listViews(Namespace namespace) {
+      return BaseViewSessionCatalog.this.listViews(context, namespace);
+    }
+
+    @Override
+    public View loadView(TableIdentifier identifier) {
+      return BaseViewSessionCatalog.this.loadView(context, identifier);
+    }
+
+    @Override
+    public boolean viewExists(TableIdentifier identifier) {
+      return BaseViewSessionCatalog.this.viewExists(context, identifier);
+    }
+
+    @Override
+    public ViewBuilder buildView(TableIdentifier identifier) {
+      return BaseViewSessionCatalog.this.buildView(context, identifier);
+    }
+
+    @Override
+    public boolean dropView(TableIdentifier identifier) {
+      return BaseViewSessionCatalog.this.dropView(context, identifier);
+    }
+
+    @Override
+    public void renameView(TableIdentifier from, TableIdentifier to) {
+      BaseViewSessionCatalog.this.renameView(context, from, to);
+    }
+
+    @Override
+    public void invalidateView(TableIdentifier identifier) {
+      BaseViewSessionCatalog.this.invalidateView(context, identifier);
+    }
+
+    @Override
+    public void initialize(String name, Map<String, String> properties) {
+      throw new UnsupportedOperationException(
+          this.getClass().getSimpleName() + " doesn't support initialization");
+    }
+  }
+}