[core] Support partition API

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

@@ -24,6 +24,10 @@
 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.JsonGetter;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+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;
@@ -48,13 +52,18 @@
  * @since 0.4.0
  */
 @Public
+@JsonIgnoreProperties(ignoreUnknown = true)
 public final class RowType extends DataType {
 
     private static final long serialVersionUID = 1L;
 
+    public static final String FILED_FIELDS = "fields";
+
     public static final String FORMAT = "ROW<%s>";
 
+    @JsonProperty(FILED_FIELDS)
     private final List<DataField> fields;
+
     private InternalRow.FieldGetter[] fieldGetters;
 
     public RowType(boolean isNullable, List<DataField> fields) {
@@ -67,14 +76,16 @@ public RowType(boolean isNullable, List<DataField> fields) {
         validateFields(fields);
     }
 
-    public RowType(List<DataField> fields) {
+    @JsonCreator
+    public RowType(@JsonProperty(FILED_FIELDS) List<DataField> fields) {
         this(true, fields);
     }
 
     public RowType copy(List<DataField> newFields) {
         return new RowType(isNullable(), newFields);
     }
 
+    @JsonGetter(FILED_FIELDS)
     public List<DataField> getFields() {
         return fields;
     }

paimon-common/src/main/java/org/apache/paimon/utils/InternalRowPartitionComputer.java

@@ -111,7 +111,7 @@ public static GenericRow convertSpecToInternalRow(
         List<String> fieldNames = partType.getFieldNames();
         for (Map.Entry<String, String> entry : spec.entrySet()) {
             Object value =
-                    defaultPartValue.equals(entry.getValue())
+                    defaultPartValue != null && defaultPartValue.equals(entry.getValue())
                             ? null
                             : castFromString(
                                     entry.getValue(), partType.getField(entry.getKey()).type());

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

@@ -94,6 +94,23 @@ public <T extends RESTResponse> T delete(String path, Map<String, String> header
         return exec(request, null);
     }
 
+    @Override
+    public <T extends RESTResponse> T delete(
+            String path, RESTRequest body, Map<String, String> headers) {
+        try {
+            RequestBody requestBody = buildRequestBody(body);
+            Request request =
+                    new Request.Builder()
+                            .url(uri + path)
+                            .delete(requestBody)
+                            .headers(Headers.of(headers))
+                            .build();
+            return exec(request, null);
+        } catch (JsonProcessingException e) {
+            throw new RESTException(e, "build request failed.");
+        }
+    }
+
     @Override
     public void close() throws IOException {
         okHttpClient.dispatcher().cancelAll();

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

@@ -20,15 +20,19 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.TableType;
+import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.catalog.CatalogUtils;
 import org.apache.paimon.catalog.Database;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.catalog.PropertyChange;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.serializer.InternalRowSerializer;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.operation.FileStoreCommit;
 import org.apache.paimon.operation.Lock;
 import org.apache.paimon.options.CatalogOptions;
 import org.apache.paimon.options.Options;
@@ -41,15 +45,19 @@
 import org.apache.paimon.rest.requests.AlterDatabaseRequest;
 import org.apache.paimon.rest.requests.AlterTableRequest;
 import org.apache.paimon.rest.requests.CreateDatabaseRequest;
+import org.apache.paimon.rest.requests.CreatePartitionRequest;
 import org.apache.paimon.rest.requests.CreateTableRequest;
+import org.apache.paimon.rest.requests.DropPartitionRequest;
 import org.apache.paimon.rest.requests.RenameTableRequest;
 import org.apache.paimon.rest.responses.AlterDatabaseResponse;
 import org.apache.paimon.rest.responses.ConfigResponse;
 import org.apache.paimon.rest.responses.CreateDatabaseResponse;
 import org.apache.paimon.rest.responses.GetDatabaseResponse;
 import org.apache.paimon.rest.responses.GetTableResponse;
 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.SuccessResponse;
 import org.apache.paimon.schema.Schema;
 import org.apache.paimon.schema.SchemaChange;
 import org.apache.paimon.schema.TableSchema;
@@ -58,10 +66,10 @@
 import org.apache.paimon.table.FileStoreTableFactory;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.object.ObjectTable;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.Preconditions;
 
-import org.apache.paimon.shade.guava30.com.google.common.annotations.VisibleForTesting;
 import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableList;
 import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -71,15 +79,21 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
 
+import static org.apache.paimon.CoreOptions.createCommitUser;
 import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemDatabase;
+import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemTable;
 import static org.apache.paimon.catalog.CatalogUtils.isSystemDatabase;
 import static org.apache.paimon.options.CatalogOptions.CASE_SENSITIVE;
+import static org.apache.paimon.rest.RESTCatalogOptions.METASTORE_PARTITIONED;
+import static org.apache.paimon.utils.InternalRowPartitionComputer.convertSpecToInternalRow;
 import static org.apache.paimon.utils.Preconditions.checkNotNull;
 import static org.apache.paimon.utils.ThreadPoolUtils.createScheduledThreadPool;
 
@@ -132,7 +146,8 @@ public RESTCatalog(CatalogContext catalogContext) {
         Map<String, String> initHeaders =
                 RESTUtil.merge(
                         configHeaders(catalogOptions.toMap()), this.catalogAuth.getHeaders());
-        Options options = new Options(fetchOptionsFromServer(initHeaders, initHeaders));
+        Options options =
+                new Options(fetchOptionsFromServer(initHeaders, catalogContext.options().toMap()));
         this.context =
                 CatalogContext.create(
                         options, catalogContext.preferIO(), catalogContext.fallbackIO());
@@ -360,17 +375,43 @@ public void dropTable(Identifier identifier, boolean ignoreIfNotExists)
     @Override
     public void createPartition(Identifier identifier, Map<String, String> partitionSpec)
             throws TableNotExistException {
-        throw new UnsupportedOperationException();
+        try {
+            CreatePartitionRequest request = new CreatePartitionRequest(identifier, partitionSpec);
+            client.post(
+                    resourcePaths.partitions(
+                            identifier.getDatabaseName(), identifier.getTableName()),
+                    request,
+                    SuccessResponse.class,
+                    headers());
+        } catch (NoSuchResourceException e) {
+            throw new TableNotExistException(identifier);
+        } catch (ForbiddenException e) {
+            throw new TableNoPermissionException(identifier, e);
+        }
     }
 
     @Override
     public void dropPartition(Identifier identifier, Map<String, String> partitions)
-            throws TableNotExistException, PartitionNotExistException {}
+            throws TableNotExistException, PartitionNotExistException {
+        checkNotSystemTable(identifier, "dropPartition");
+        dropPartitionMetadata(identifier, partitions);
+        Table table = getTable(identifier);
+        if (table != null) {
+            cleanPartitionsInFileSystem(table, partitions);
+        } else {
+            throw new TableNotExistException(identifier);
+        }
+    }
 
     @Override
     public List<PartitionEntry> listPartitions(Identifier identifier)
             throws TableNotExistException {
-        throw new UnsupportedOperationException();
+        boolean whetherSupportListPartitions = context.options().get(METASTORE_PARTITIONED);
+        if (whetherSupportListPartitions) {
+            return listPartitionsFromServer(identifier);
+        } else {
+            return getTable(identifier).newReadBuilder().newScan().listPartitionEntries();
+        }
     }
 
     @Override
@@ -420,6 +461,56 @@ Table getDataOrFormatTable(Identifier identifier) throws TableNotExistException
         return table;
     }
 
+    @VisibleForTesting
+    public List<PartitionEntry> listPartitionsFromServer(Identifier identifier)
+            throws TableNotExistException {
+        try {
+            ListPartitionsResponse response =
+                    client.get(
+                            resourcePaths.partitions(
+                                    identifier.getDatabaseName(), identifier.getTableName()),
+                            ListPartitionsResponse.class,
+                            headers());
+            if (response != null && response.getPartitions() != null) {
+                return response.getPartitions().stream()
+                        .map(this::convertToPartitionEntry)
+                        .collect(Collectors.toList());
+            } else {
+                return Collections.emptyList();
+            }
+        } catch (NoSuchResourceException e) {
+            throw new TableNotExistException(identifier);
+        } catch (ForbiddenException e) {
+            throw new TableNoPermissionException(identifier, e);
+        }
+    }
+
+    @VisibleForTesting
+    PartitionEntry convertToPartitionEntry(ListPartitionsResponse.Partition partition) {
+        InternalRowSerializer serializer = new InternalRowSerializer(partition.getPartitionType());
+        GenericRow row =
+                convertSpecToInternalRow(partition.getSpec(), partition.getPartitionType(), null);
+        return new PartitionEntry(
+                serializer.toBinaryRow(row).copy(),
+                partition.getRecordCount(),
+                partition.getFileSizeInBytes(),
+                partition.getFileCount(),
+                partition.getLastFileCreationTime());
+    }
+
+    @VisibleForTesting
+    void cleanPartitionsInFileSystem(Table table, Map<String, String> partitions) {
+        FileStoreTable fileStoreTable = (FileStoreTable) table;
+        try (FileStoreCommit commit =
+                fileStoreTable
+                        .store()
+                        .newCommit(
+                                createCommitUser(fileStoreTable.coreOptions().toConfiguration()))) {
+            commit.dropPartitions(
+                    Collections.singletonList(partitions), BatchWriteBuilder.COMMIT_IDENTIFIER);
+        }
+    }
+
     protected GetTableResponse getTableResponse(Identifier identifier)
             throws TableNotExistException {
         try {
@@ -434,6 +525,23 @@ protected GetTableResponse getTableResponse(Identifier identifier)
         }
     }
 
+    protected SuccessResponse dropPartitionMetadata(
+            Identifier identifier, Map<String, String> partitions)
+            throws TableNoPermissionException {
+        try {
+            DropPartitionRequest request = new DropPartitionRequest(partitions);
+            return client.delete(
+                    resourcePaths.partitions(
+                            identifier.getDatabaseName(), identifier.getTableName()),
+                    request,
+                    headers());
+        } catch (NoSuchResourceException ignore) {
+            return new SuccessResponse();
+        } catch (ForbiddenException e) {
+            throw new TableNoPermissionException(identifier, e);
+        }
+    }
+
     private static Map<String, String> configHeaders(Map<String, String> properties) {
         return RESTUtil.extractPrefixMap(properties, "header.");
     }

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

@@ -71,4 +71,10 @@ public class RESTCatalogOptions {
                     .stringType()
                     .noDefaultValue()
                     .withDescription("REST Catalog auth token provider path.");
+
+    public static final ConfigOption<Boolean> METASTORE_PARTITIONED =
+            ConfigOptions.key("metastore-partitioned")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("REST Catalog Server whether support list partitions.");
 }

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

@@ -30,4 +30,6 @@ <T extends RESTResponse> T post(
             String path, RESTRequest body, Class<T> responseType, Map<String, String> headers);
 
     <T extends RESTResponse> T delete(String path, Map<String, String> headers);
+
+    <T extends RESTResponse> T delete(String path, RESTRequest body, Map<String, String> headers);
 }

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

@@ -82,4 +82,15 @@ public String renameTable(String databaseName, String tableName) {
                 .add("rename")
                 .toString();
     }
+
+    public String partitions(String databaseName, String tableName) {
+        return SLASH.add("v1")
+                .add(prefix)
+                .add("databases")
+                .add(databaseName)
+                .add("tables")
+                .add(tableName)
+                .add("partitions")
+                .toString();
+    }
 }

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

@@ -0,0 +1,61 @@
+/*
+ * 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.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;
+
+import java.util.Map;
+
+/** Request for creating partition. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CreatePartitionRequest implements RESTRequest {
+
+    private static final String FIELD_IDENTIFIER = "identifier";
+    private static final String FIELD_PARTITION_SPEC = "partitionSpec";
+
+    @JsonProperty(FIELD_IDENTIFIER)
+    private final Identifier identifier;
+
+    @JsonProperty(FIELD_PARTITION_SPEC)
+    private final Map<String, String> partitionSpec;
+
+    @JsonCreator
+    public CreatePartitionRequest(
+            @JsonProperty(FIELD_IDENTIFIER) Identifier identifier,
+            @JsonProperty(FIELD_PARTITION_SPEC) Map<String, String> partitionSpec) {
+        this.identifier = identifier;
+        this.partitionSpec = partitionSpec;
+    }
+
+    @JsonGetter(FIELD_IDENTIFIER)
+    public Identifier getIdentifier() {
+        return identifier;
+    }
+
+    @JsonGetter(FIELD_PARTITION_SPEC)
+    public Map<String, String> getPartitionSpec() {
+        return partitionSpec;
+    }
+}