[core] Support view API in Rest catalog

paimon-common/src/main/java/org/apache/paimon/types/RowType.java

@@ -24,6 +24,8 @@
 import org.apache.paimon.utils.Preconditions;
 import org.apache.paimon.utils.StringUtils;
 
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.JsonGenerator;
 
 import java.io.IOException;
@@ -52,6 +54,8 @@ public final class RowType extends DataType {
 
     private static final long serialVersionUID = 1L;
 
+    private static final String FIELD_FIELDS = "fields";
+
     public static final String FORMAT = "ROW<%s>";
 
     private final List<DataField> fields;
@@ -67,7 +71,8 @@ public RowType(boolean isNullable, List<DataField> fields) {
         validateFields(fields);
     }
 
-    public RowType(List<DataField> fields) {
+    @JsonCreator
+    public RowType(@JsonProperty(FIELD_FIELDS) List<DataField> fields) {
         this(true, fields);
     }
 

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -45,6 +45,7 @@
 import org.apache.paimon.rest.requests.CreateDatabaseRequest;
 import org.apache.paimon.rest.requests.CreatePartitionsRequest;
 import org.apache.paimon.rest.requests.CreateTableRequest;
+import org.apache.paimon.rest.requests.CreateViewRequest;
 import org.apache.paimon.rest.requests.DropPartitionsRequest;
 import org.apache.paimon.rest.requests.MarkDonePartitionsRequest;
 import org.apache.paimon.rest.requests.RenameTableRequest;
@@ -55,16 +56,22 @@
 import org.apache.paimon.rest.responses.ErrorResponseResourceType;
 import org.apache.paimon.rest.responses.GetDatabaseResponse;
 import org.apache.paimon.rest.responses.GetTableResponse;
+import org.apache.paimon.rest.responses.GetViewResponse;
 import org.apache.paimon.rest.responses.ListDatabasesResponse;
 import org.apache.paimon.rest.responses.ListPartitionsResponse;
 import org.apache.paimon.rest.responses.ListTablesResponse;
+import org.apache.paimon.rest.responses.ListViewsResponse;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.Pair;
+import org.apache.paimon.view.View;
+import org.apache.paimon.view.ViewImpl;
+import org.apache.paimon.view.ViewSchema;
 
 import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
 
@@ -290,8 +297,7 @@ public boolean commitSnapshot(Identifier identifier, Snapshot snapshot) {
         CommitTableRequest request = new CommitTableRequest(identifier, snapshot);
         CommitTableResponse response =
                 client.post(
-                        resourcePaths.commitTable(
-                                identifier.getDatabaseName(), identifier.getTableName()),
+                        resourcePaths.commitTable(identifier.getDatabaseName()),
                         request,
                         CommitTableResponse.class,
                         headers());
@@ -325,11 +331,7 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx
             checkNotSystemTable(identifier, "createTable");
             validateAutoCreateClose(schema.options());
             CreateTableRequest request = new CreateTableRequest(identifier, schema);
-            client.post(
-                    resourcePaths.tables(identifier.getDatabaseName()),
-                    request,
-                    GetTableResponse.class,
-                    headers());
+            client.post(resourcePaths.tables(identifier.getDatabaseName()), request, headers());
         } catch (AlreadyExistsException e) {
             if (!ignoreIfExists) {
                 throw new TableAlreadyExistException(identifier);
@@ -353,13 +355,8 @@ public void renameTable(Identifier fromTable, Identifier toTable, boolean ignore
         checkNotSystemTable(fromTable, "renameTable");
         checkNotSystemTable(toTable, "renameTable");
         try {
-            RenameTableRequest request = new RenameTableRequest(toTable);
-            client.post(
-                    resourcePaths.renameTable(
-                            fromTable.getDatabaseName(), fromTable.getTableName()),
-                    request,
-                    GetTableResponse.class,
-                    headers());
+            RenameTableRequest request = new RenameTableRequest(fromTable, toTable);
+            client.post(resourcePaths.renameTable(fromTable.getDatabaseName()), request, headers());
         } catch (NoSuchResourceException e) {
             if (!ignoreIfNotExists) {
                 throw new TableNotExistException(fromTable);
@@ -381,7 +378,6 @@ public void alterTable(
             client.post(
                     resourcePaths.table(identifier.getDatabaseName(), identifier.getTableName()),
                     request,
-                    GetTableResponse.class,
                     headers());
         } catch (NoSuchResourceException e) {
             if (!ignoreIfNotExists) {
@@ -532,6 +528,88 @@ public List<Partition> listPartitions(Identifier identifier) throws TableNotExis
         return response.getPartitions();
     }
 
+    @Override
+    public View getView(Identifier identifier) throws ViewNotExistException {
+        try {
+            GetViewResponse response =
+                    client.get(
+                            resourcePaths.view(
+                                    identifier.getDatabaseName(), identifier.getTableName()),
+                            GetViewResponse.class,
+                            headers());
+            return new ViewImpl(
+                    identifier,
+                    response.getSchema().rowType(),
+                    response.getSchema().query(),
+                    response.getSchema().comment(),
+                    response.getSchema().options());
+        } catch (NoSuchResourceException e) {
+            throw new ViewNotExistException(identifier);
+        }
+    }
+
+    @Override
+    public void dropView(Identifier identifier, boolean ignoreIfNotExists)
+            throws ViewNotExistException {
+        try {
+            client.delete(
+                    resourcePaths.view(identifier.getDatabaseName(), identifier.getTableName()),
+                    headers());
+        } catch (NoSuchResourceException e) {
+            if (!ignoreIfNotExists) {
+                throw new ViewNotExistException(identifier);
+            }
+        }
+    }
+
+    @Override
+    public void createView(Identifier identifier, View view, boolean ignoreIfExists)
+            throws ViewAlreadyExistException, DatabaseNotExistException {
+        try {
+            ViewSchema schema =
+                    new ViewSchema(
+                            new RowType(view.rowType().getFields()),
+                            view.options(),
+                            view.comment().orElse(null),
+                            view.query());
+            CreateViewRequest request = new CreateViewRequest(identifier, schema);
+            client.post(resourcePaths.views(identifier.getDatabaseName()), request, headers());
+        } catch (NoSuchResourceException e) {
+            throw new DatabaseNotExistException(identifier.getDatabaseName());
+        } catch (AlreadyExistsException e) {
+            if (!ignoreIfExists) {
+                throw new ViewAlreadyExistException(identifier);
+            }
+        }
+    }
+
+    @Override
+    public List<String> listViews(String databaseName) throws DatabaseNotExistException {
+        try {
+            ListViewsResponse response =
+                    client.get(
+                            resourcePaths.views(databaseName), ListViewsResponse.class, headers());
+            return response.getViews();
+        } catch (NoSuchResourceException e) {
+            throw new DatabaseNotExistException(databaseName);
+        }
+    }
+
+    @Override
+    public void renameView(Identifier fromView, Identifier toView, boolean ignoreIfNotExists)
+            throws ViewNotExistException, ViewAlreadyExistException {
+        try {
+            RenameTableRequest request = new RenameTableRequest(fromView, toView);
+            client.post(resourcePaths.renameView(fromView.getDatabaseName()), request, headers());
+        } catch (NoSuchResourceException e) {
+            if (!ignoreIfNotExists) {
+                throw new ViewNotExistException(fromView);
+            }
+        } catch (AlreadyExistsException e) {
+            throw new ViewAlreadyExistException(toView);
+        }
+    }
+
     @Override
     public boolean caseSensitive() {
         return options.getOptional(CASE_SENSITIVE).orElse(true);

paimon-core/src/main/java/org/apache/paimon/rest/ResourcePaths.java

@@ -62,12 +62,12 @@ public String table(String databaseName, String tableName) {
         return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, tableName);
     }
 
-    public String renameTable(String databaseName, String tableName) {
-        return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, tableName, "rename");
+    public String renameTable(String databaseName) {
+        return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, "rename");
     }
 
-    public String commitTable(String databaseName, String tableName) {
-        return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, tableName, "commit");
+    public String commitTable(String databaseName) {
+        return SLASH.join(V1, prefix, DATABASES, databaseName, TABLES, "commit");
     }
 
     public String partitions(String databaseName, String tableName) {
@@ -88,4 +88,16 @@ public String markDonePartitions(String databaseName, String tableName) {
         return SLASH.join(
                 V1, prefix, DATABASES, databaseName, TABLES, tableName, "partitions", "mark");
     }
+
+    public String views(String databaseName) {
+        return SLASH.join(V1, prefix, DATABASES, databaseName, "views");
+    }
+
+    public String view(String databaseName, String viewName) {
+        return SLASH.join(V1, prefix, DATABASES, databaseName, "views", viewName);
+    }
+
+    public String renameView(String databaseName) {
+        return SLASH.join(V1, prefix, DATABASES, databaseName, "views", "rename");
+    }
 }

paimon-core/src/main/java/org/apache/paimon/rest/requests/CreateViewRequest.java

@@ -0,0 +1,60 @@
+/*
+ * 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.paimon.rest.requests;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.rest.RESTRequest;
+import org.apache.paimon.view.ViewSchema;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+/** Request for creating view. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CreateViewRequest implements RESTRequest {
+
+    private static final String FIELD_IDENTIFIER = "identifier";
+    private static final String FIELD_SCHEMA = "schema";
+
+    @JsonProperty(FIELD_IDENTIFIER)
+    private final Identifier identifier;
+
+    @JsonProperty(FIELD_SCHEMA)
+    private final ViewSchema schema;
+
+    @JsonCreator
+    public CreateViewRequest(
+            @JsonProperty(FIELD_IDENTIFIER) Identifier identifier,
+            @JsonProperty(FIELD_SCHEMA) ViewSchema schema) {
+        this.schema = schema;
+        this.identifier = identifier;
+    }
+
+    @JsonGetter(FIELD_IDENTIFIER)
+    public Identifier getIdentifier() {
+        return identifier;
+    }
+
+    @JsonGetter(FIELD_SCHEMA)
+    public ViewSchema getSchema() {
+        return schema;
+    }
+}

paimon-core/src/main/java/org/apache/paimon/rest/requests/RenameTableRequest.java

@@ -26,22 +26,34 @@
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
 
-/** Request for renaming table. */
+/** Request for renaming. */
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class RenameTableRequest implements RESTRequest {
 
-    private static final String FIELD_NEW_IDENTIFIER_NAME = "newIdentifier";
+    private static final String FIELD_SOURCE = "source";
+    private static final String FIELD_DESTINATION = "destination";
 
-    @JsonProperty(FIELD_NEW_IDENTIFIER_NAME)
-    private final Identifier newIdentifier;
+    @JsonProperty(FIELD_SOURCE)
+    private final Identifier source;
+
+    @JsonProperty(FIELD_DESTINATION)
+    private final Identifier destination;
 
     @JsonCreator
-    public RenameTableRequest(@JsonProperty(FIELD_NEW_IDENTIFIER_NAME) Identifier newIdentifier) {
-        this.newIdentifier = newIdentifier;
+    public RenameTableRequest(
+            @JsonProperty(FIELD_SOURCE) Identifier source,
+            @JsonProperty(FIELD_DESTINATION) Identifier destination) {
+        this.source = source;
+        this.destination = destination;
+    }
+
+    @JsonGetter(FIELD_DESTINATION)
+    public Identifier getDestination() {
+        return destination;
     }
 
-    @JsonGetter(FIELD_NEW_IDENTIFIER_NAME)
-    public Identifier getNewIdentifier() {
-        return newIdentifier;
+    @JsonGetter(FIELD_SOURCE)
+    public Identifier getSource() {
+        return source;
     }
 }

paimon-core/src/main/java/org/apache/paimon/rest/responses/ErrorResponseResourceType.java

@@ -23,4 +23,5 @@ public enum ErrorResponseResourceType {
     DATABASE,
     TABLE,
     COLUMN,
+    VIEW
 }