REST: Implement register view for REST catalog

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

@@ -106,6 +106,18 @@ default boolean viewExists(TableIdentifier identifier) {
    */
   default void invalidateView(TableIdentifier identifier) {}
 
+  /**
+   * Register a view with the catalog if it does not exist.
+   *
+   * @param identifier a view identifier
+   * @param metadataFileLocation the location of a metadata file
+   * @return a View instance
+   * @throws AlreadyExistsException if the view already exists in the catalog.
+   */
+  default View registerView(TableIdentifier identifier, String metadataFileLocation) {
+    throw new UnsupportedOperationException("Registering views is not supported");
+  }
+
   /**
    * Initialize a view catalog given a custom name and a map of catalog properties.
    *

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

@@ -106,6 +106,20 @@ default boolean viewExists(SessionCatalog.SessionContext context, TableIdentifie
    */
   default void invalidateView(SessionCatalog.SessionContext context, TableIdentifier identifier) {}
 
+  /**
+   * Register a view if it does not exist.
+   *
+   * @param context session context
+   * @param ident a view identifier
+   * @param metadataFileLocation the location of a metadata file
+   * @return a View instance
+   * @throws AlreadyExistsException if the view already exists in the catalog.
+   */
+  default View registerView(
+      SessionCatalog.SessionContext context, TableIdentifier ident, String metadataFileLocation) {
+    throw new UnsupportedOperationException("Registering views is not supported");
+  }
+
   /**
    * Initialize a view catalog given a custom name and a map of catalog properties.
    *

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

@@ -83,6 +83,11 @@ public void invalidateView(TableIdentifier identifier) {
       BaseViewSessionCatalog.this.invalidateView(context, identifier);
     }
 
+    @Override
+    public View registerView(TableIdentifier identifier, String metadataFileLocation) {
+      return BaseViewSessionCatalog.this.registerView(context, identifier, metadataFileLocation);
+    }
+
     @Override
     public void initialize(String name, Map<String, String> properties) {
       throw new UnsupportedOperationException(

core/src/main/java/org/apache/iceberg/rest/CatalogHandlers.java

@@ -77,6 +77,7 @@
 import org.apache.iceberg.rest.requests.FetchScanTasksRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.RegisterTableRequest;
+import org.apache.iceberg.rest.requests.RegisterViewRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
@@ -595,6 +596,18 @@ public static void dropView(ViewCatalog catalog, TableIdentifier viewIdentifier)
     }
   }
 
+  public static LoadViewResponse registerView(
+      ViewCatalog catalog, Namespace namespace, RegisterViewRequest request) {
+    request.validate();
+
+    TableIdentifier identifier = TableIdentifier.of(namespace, request.name());
+    View view = catalog.registerView(identifier, request.metadataLocation());
+    return ImmutableLoadViewResponse.builder()
+        .metadata(asBaseView(view).operations().current())
+        .metadataLocation(request.metadataLocation())
+        .build();
+  }
+
   static ViewMetadata commit(ViewOperations ops, UpdateTableRequest request) {
     AtomicBoolean isRetry = new AtomicBoolean(false);
     try {

core/src/main/java/org/apache/iceberg/rest/Endpoint.java

@@ -86,6 +86,8 @@ public class Endpoint {
   public static final Endpoint V1_DELETE_VIEW = Endpoint.create("DELETE", ResourcePaths.V1_VIEW);
   public static final Endpoint V1_RENAME_VIEW =
       Endpoint.create("POST", ResourcePaths.V1_VIEW_RENAME);
+  public static final Endpoint V1_REGISTER_VIEW =
+      Endpoint.create("POST", ResourcePaths.V1_VIEW_REGISTER);
 
   private static final Splitter ENDPOINT_SPLITTER = Splitter.on(" ");
   private static final Joiner ENDPOINT_JOINER = Joiner.on(" ");

core/src/main/java/org/apache/iceberg/rest/RESTCatalog.java

@@ -325,4 +325,9 @@ public boolean viewExists(TableIdentifier identifier) {
   public void invalidateView(TableIdentifier identifier) {
     viewSessionCatalog.invalidateView(identifier);
   }
+
+  @Override
+  public View registerView(TableIdentifier identifier, String metadataFileLocation) {
+    return viewSessionCatalog.registerView(identifier, metadataFileLocation);
+  }
 }

core/src/main/java/org/apache/iceberg/rest/RESTSerializers.java

@@ -52,11 +52,14 @@
 import org.apache.iceberg.rest.requests.FetchScanTasksRequestParser;
 import org.apache.iceberg.rest.requests.ImmutableCreateViewRequest;
 import org.apache.iceberg.rest.requests.ImmutableRegisterTableRequest;
+import org.apache.iceberg.rest.requests.ImmutableRegisterViewRequest;
 import org.apache.iceberg.rest.requests.ImmutableReportMetricsRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequestParser;
 import org.apache.iceberg.rest.requests.RegisterTableRequest;
 import org.apache.iceberg.rest.requests.RegisterTableRequestParser;
+import org.apache.iceberg.rest.requests.RegisterViewRequest;
+import org.apache.iceberg.rest.requests.RegisterViewRequestParser;
 import org.apache.iceberg.rest.requests.ReportMetricsRequest;
 import org.apache.iceberg.rest.requests.ReportMetricsRequestParser;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
@@ -131,6 +134,11 @@ public static void registerAll(ObjectMapper mapper) {
         .addSerializer(ImmutableLoadViewResponse.class, new LoadViewResponseSerializer<>())
         .addDeserializer(LoadViewResponse.class, new LoadViewResponseDeserializer<>())
         .addDeserializer(ImmutableLoadViewResponse.class, new LoadViewResponseDeserializer<>())
+        .addSerializer(RegisterViewRequest.class, new RegisterViewRequestSerializer<>())
+        .addDeserializer(RegisterViewRequest.class, new RegisterViewRequestDeserializer<>())
+        .addSerializer(ImmutableRegisterViewRequest.class, new RegisterViewRequestSerializer<>())
+        .addDeserializer(
+            ImmutableRegisterViewRequest.class, new RegisterViewRequestDeserializer<>())
         .addSerializer(ConfigResponse.class, new ConfigResponseSerializer<>())
         .addDeserializer(ConfigResponse.class, new ConfigResponseDeserializer<>())
         .addSerializer(LoadTableResponse.class, new LoadTableResponseSerializer<>())
@@ -444,6 +452,24 @@ public T deserialize(JsonParser p, DeserializationContext context) throws IOExce
     }
   }
 
+  public static class RegisterViewRequestSerializer<T extends RegisterViewRequest>
+      extends JsonSerializer<T> {
+    @Override
+    public void serialize(T request, JsonGenerator gen, SerializerProvider serializers)
+        throws IOException {
+      RegisterViewRequestParser.toJson(request, gen);
+    }
+  }
+
+  public static class RegisterViewRequestDeserializer<T extends RegisterViewRequest>
+      extends JsonDeserializer<T> {
+    @Override
+    public T deserialize(JsonParser p, DeserializationContext context) throws IOException {
+      JsonNode jsonNode = p.getCodec().readTree(p);
+      return (T) RegisterViewRequestParser.fromJson(jsonNode);
+    }
+  }
+
   static class ConfigResponseSerializer<T extends ConfigResponse> extends JsonSerializer<T> {
     @Override
     public void serialize(T request, JsonGenerator gen, SerializerProvider serializers)

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -77,7 +77,9 @@
 import org.apache.iceberg.rest.requests.CreateViewRequest;
 import org.apache.iceberg.rest.requests.ImmutableCreateViewRequest;
 import org.apache.iceberg.rest.requests.ImmutableRegisterTableRequest;
+import org.apache.iceberg.rest.requests.ImmutableRegisterViewRequest;
 import org.apache.iceberg.rest.requests.RegisterTableRequest;
+import org.apache.iceberg.rest.requests.RegisterViewRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
@@ -1366,6 +1368,48 @@ public void renameView(SessionContext context, TableIdentifier from, TableIdenti
         .post(paths.renameView(), request, null, mutationHeaders, ErrorHandlers.viewErrorHandler());
   }
 
+  @Override
+  public View registerView(
+      SessionContext context, TableIdentifier ident, String metadataFileLocation) {
+    Endpoint.check(endpoints, Endpoint.V1_REGISTER_VIEW);
+    checkViewIdentifierIsValid(ident);
+
+    Preconditions.checkArgument(
+        metadataFileLocation != null && !metadataFileLocation.isEmpty(),
+        "Invalid metadata file location: %s",
+        metadataFileLocation);
+
+    RegisterViewRequest request =
+        ImmutableRegisterViewRequest.builder()
+            .name(ident.name())
+            .metadataLocation(metadataFileLocation)
+            .build();
+
+    AuthSession contextualSession = authManager.contextualSession(context, catalogAuth);
+    LoadViewResponse response =
+        client
+            .withAuthSession(contextualSession)
+            .post(
+                paths.registerView(ident.namespace()),
+                request,
+                LoadViewResponse.class,
+                mutationHeaders,
+                ErrorHandlers.viewErrorHandler());
+
+    AuthSession tableSession =
+        authManager.tableSession(ident, response.config(), contextualSession);
+    RESTViewOperations ops =
+        newViewOps(
+            client.withAuthSession(tableSession),
+            paths.view(ident),
+            Map::of,
+            mutationHeaders,
+            response.metadata(),
+            endpoints);
+
+    return new BaseView(ops, ViewUtil.fullViewName(name(), ident));
+  }
+
   private class RESTViewBuilder implements ViewBuilder {
     private final SessionContext context;
     private final TableIdentifier identifier;

core/src/main/java/org/apache/iceberg/rest/ResourcePaths.java

@@ -49,6 +49,7 @@ public class ResourcePaths {
   public static final String V1_VIEWS = "/v1/{prefix}/namespaces/{namespace}/views";
   public static final String V1_VIEW = "/v1/{prefix}/namespaces/{namespace}/views/{view}";
   public static final String V1_VIEW_RENAME = "/v1/{prefix}/views/rename";
+  public static final String V1_VIEW_REGISTER = "/v1/{prefix}/namespaces/{namespace}/register-view";
 
   public static ResourcePaths forCatalogProperties(Map<String, String> properties) {
     return new ResourcePaths(
@@ -151,6 +152,10 @@ public String renameView() {
     return SLASH.join("v1", prefix, "views", "rename");
   }
 
+  public String registerView(Namespace ns) {
+    return SLASH.join("v1", prefix, "namespaces", pathEncode(ns), "register-view");
+  }
+
   public String planTableScan(TableIdentifier ident) {
     return SLASH.join(
         "v1",

core/src/main/java/org/apache/iceberg/rest/requests/RegisterViewRequest.java

@@ -0,0 +1,35 @@
+/*
+ * 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.rest.requests;
+
+import org.apache.iceberg.rest.RESTRequest;
+import org.immutables.value.Value;
+
+@Value.Immutable
+public interface RegisterViewRequest extends RESTRequest {
+
+  String name();
+
+  String metadataLocation();
+
+  @Override
+  default void validate() {
+    // nothing to validate as it's not possible to create an invalid instance
+  }
+}

core/src/main/java/org/apache/iceberg/rest/requests/RegisterViewRequestParser.java

@@ -0,0 +1,69 @@
+/*
+ * 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.rest.requests;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+public class RegisterViewRequestParser {
+
+  private static final String NAME = "name";
+  private static final String METADATA_LOCATION = "metadata-location";
+
+  private RegisterViewRequestParser() {}
+
+  public static String toJson(RegisterViewRequest request) {
+    return toJson(request, false);
+  }
+
+  public static String toJson(RegisterViewRequest request, boolean pretty) {
+    return JsonUtil.generate(gen -> toJson(request, gen), pretty);
+  }
+
+  public static void toJson(RegisterViewRequest request, JsonGenerator gen) throws IOException {
+    Preconditions.checkArgument(null != request, "Invalid register view request: null");
+
+    gen.writeStartObject();
+
+    gen.writeStringField(NAME, request.name());
+    gen.writeStringField(METADATA_LOCATION, request.metadataLocation());
+
+    gen.writeEndObject();
+  }
+
+  public static RegisterViewRequest fromJson(String json) {
+    return JsonUtil.parse(json, RegisterViewRequestParser::fromJson);
+  }
+
+  public static RegisterViewRequest fromJson(JsonNode json) {
+    Preconditions.checkArgument(
+        null != json, "Cannot parse register view request from null object");
+
+    String name = JsonUtil.getString(NAME, json);
+    String metadataLocation = JsonUtil.getString(METADATA_LOCATION, json);
+
+    return ImmutableRegisterViewRequest.builder()
+        .name(name)
+        .metadataLocation(metadataLocation)
+        .build();
+  }
+}

core/src/main/java/org/apache/iceberg/view/BaseMetastoreViewCatalog.java

@@ -31,6 +31,7 @@
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.NoSuchViewException;
+import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -298,4 +299,28 @@ public Transaction replaceTransaction() {
       return super.replaceTransaction();
     }
   }
+
+  @Override
+  public View registerView(TableIdentifier identifier, String metadataFileLocation) {
+    Preconditions.checkArgument(
+        identifier != null && isValidIdentifier(identifier), "Invalid identifier: %s", identifier);
+    Preconditions.checkArgument(
+        metadataFileLocation != null && !metadataFileLocation.isEmpty(),
+        "Cannot register an empty metadata file location as a view");
+
+    if (viewExists(identifier)) {
+      throw new AlreadyExistsException("View already exists: %s", identifier);
+    }
+
+    if (tableExists(identifier)) {
+      throw new AlreadyExistsException("Table with same name already exists: %s", identifier);
+    }
+
+    ViewOperations ops = newViewOps(identifier);
+    InputFile metadataFile = ((BaseViewOperations) ops).io().newInputFile(metadataFileLocation);
+    ViewMetadata metadata = ViewMetadataParser.read(metadataFile);
+    ops.commit(null, metadata);
+
+    return new BaseView(ops, ViewUtil.fullViewName(name(), identifier));
+  }
 }

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

@@ -69,6 +69,7 @@
 import org.apache.iceberg.rest.requests.FetchScanTasksRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
 import org.apache.iceberg.rest.requests.RegisterTableRequest;
+import org.apache.iceberg.rest.requests.RegisterViewRequest;
 import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.ReportMetricsRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
@@ -474,6 +475,17 @@ public <T extends RESTResponse> T handleRequest(
           break;
         }
 
+      case REGISTER_VIEW:
+        {
+          if (null != asViewCatalog) {
+            Namespace namespace = namespaceFromPathVars(vars);
+            RegisterViewRequest request = castRequest(RegisterViewRequest.class, body);
+            return castResponse(
+                responseType, CatalogHandlers.registerView(asViewCatalog, namespace, request));
+          }
+          break;
+        }
+
       default:
         if (responseType == OAuthTokenResponse.class) {
           return castResponse(responseType, handleOAuthRequest(body));