[core][rest] Support catalog-managed Format Table partitions

docs/static/rest-catalog-open-api.yaml

@@ -983,6 +983,90 @@ paths:
           $ref: '#/components/responses/TableNotExistErrorResponse'
         "500":
           $ref: '#/components/responses/ServerErrorResponse'
+    post:
+      tags:
+        - partition
+      summary: Create partitions
+      operationId: createPartitions
+      parameters:
+        - name: prefix
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: database
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: table
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CreatePartitionsRequest'
+      responses:
+        "200":
+          description: Partitions created or found to exist
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/CreatePartitionsResponse'
+        "401":
+          $ref: '#/components/responses/UnauthorizedErrorResponse'
+        "404":
+          $ref: '#/components/responses/TableNotExistErrorResponse'
+        "409":
+          $ref: '#/components/responses/ResourceAlreadyExistErrorResponse'
+        "500":
+          $ref: '#/components/responses/ServerErrorResponse'
+  /v1/{prefix}/databases/{database}/tables/{table}/partitions/drop:
+    post:
+      tags:
+        - partition
+      summary: Drop partitions
+      description: Unregisters partitions from the catalog. The server never deletes data files; managed format table data deletion is performed by the client afterwards.
+      operationId: dropPartitions
+      parameters:
+        - name: prefix
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: database
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: table
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/DropPartitionsRequest'
+      responses:
+        "200":
+          description: Partitions dropped or found missing
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/DropPartitionsResponse'
+        "401":
+          $ref: '#/components/responses/UnauthorizedErrorResponse'
+        "404":
+          $ref: '#/components/responses/TableNotExistErrorResponse'
+        "500":
+          $ref: '#/components/responses/ServerErrorResponse'
   /v1/{prefix}/databases/{database}/tables/{table}/partitions/mark:
     post:
       tags:
@@ -2352,11 +2436,68 @@ components:
             type: string
     DropPartitionsRequest:
       type: object
+      required:
+        - partitionSpecs
+      properties:
+        partitionSpecs:
+          type: array
+          items:
+            type: object
+            additionalProperties:
+              type: string
+        ignoreIfNotExists:
+          type: boolean
+          default: true
+    DropPartitionsResponse:
+      type: object
+      required:
+        - dropped
+        - missing
       properties:
-        specs:
+        dropped:
           type: array
           items:
             type: object
+            additionalProperties:
+              type: string
+        missing:
+          type: array
+          items:
+            type: object
+            additionalProperties:
+              type: string
+    CreatePartitionsRequest:
+      type: object
+      required:
+        - partitionSpecs
+      properties:
+        partitionSpecs:
+          type: array
+          items:
+            type: object
+            additionalProperties:
+              type: string
+        ignoreIfExists:
+          type: boolean
+          default: true
+    CreatePartitionsResponse:
+      type: object
+      required:
+        - created
+        - existed
+      properties:
+        created:
+          type: array
+          items:
+            type: object
+            additionalProperties:
+              type: string
+        existed:
+          type: array
+          items:
+            type: object
+            additionalProperties:
+              type: string
     AlterTableRequest:
       type: object
       properties:
@@ -2398,7 +2539,7 @@ components:
         resourceType:
           type: string
           nullable: true
-          enum: [ "DATABASE", "TABLE", "COLUMN", "SNAPSHOT", "BRANCH", "TAG", "VIEW", "DIALECT", "UNKNOWN" ]
+          enum: [ "DATABASE", "TABLE", "PARTITION", "COLUMN", "SNAPSHOT", "BRANCH", "TAG", "VIEW", "DIALECT", "UNKNOWN" ]
         resourceName:
           type: string
           nullable: true

paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java

@@ -42,9 +42,11 @@
 import org.apache.paimon.rest.requests.CreateBranchRequest;
 import org.apache.paimon.rest.requests.CreateDatabaseRequest;
 import org.apache.paimon.rest.requests.CreateFunctionRequest;
+import org.apache.paimon.rest.requests.CreatePartitionsRequest;
 import org.apache.paimon.rest.requests.CreateTableRequest;
 import org.apache.paimon.rest.requests.CreateTagRequest;
 import org.apache.paimon.rest.requests.CreateViewRequest;
+import org.apache.paimon.rest.requests.DropPartitionsRequest;
 import org.apache.paimon.rest.requests.ForwardBranchRequest;
 import org.apache.paimon.rest.requests.ListPartitionsByNamesRequest;
 import org.apache.paimon.rest.requests.MarkDonePartitionsRequest;
@@ -58,6 +60,8 @@
 import org.apache.paimon.rest.responses.AuthTableQueryResponse;
 import org.apache.paimon.rest.responses.CommitTableResponse;
 import org.apache.paimon.rest.responses.ConfigResponse;
+import org.apache.paimon.rest.responses.CreatePartitionsResponse;
+import org.apache.paimon.rest.responses.DropPartitionsResponse;
 import org.apache.paimon.rest.responses.ErrorResponse;
 import org.apache.paimon.rest.responses.GetDatabaseResponse;
 import org.apache.paimon.rest.responses.GetFunctionResponse;
@@ -855,6 +859,37 @@ public void markDonePartitions(Identifier identifier, List<Map<String, String>>
                 restAuthFunction);
     }
 
+    /** Create partitions for table, ignoring partitions which already exist. */
+    public CreatePartitionsResponse createPartitions(
+            Identifier identifier, List<Map<String, String>> partitions) {
+        return createPartitions(identifier, partitions, true);
+    }
+
+    /** Create partitions for table. */
+    public CreatePartitionsResponse createPartitions(
+            Identifier identifier, List<Map<String, String>> partitions, boolean ignoreIfExists) {
+        CreatePartitionsRequest request = new CreatePartitionsRequest(partitions, ignoreIfExists);
+        return client.post(
+                resourcePaths.partitions(identifier.getDatabaseName(), identifier.getObjectName()),
+                request,
+                CreatePartitionsResponse.class,
+                restAuthFunction);
+    }
+
+    /** Drop (unregister) partitions for table; the server never deletes data files. */
+    public DropPartitionsResponse dropPartitions(
+            Identifier identifier,
+            List<Map<String, String>> partitions,
+            boolean ignoreIfNotExists) {
+        DropPartitionsRequest request = new DropPartitionsRequest(partitions, ignoreIfNotExists);
+        return client.post(
+                resourcePaths.dropPartitions(
+                        identifier.getDatabaseName(), identifier.getObjectName()),
+                request,
+                DropPartitionsResponse.class,
+                restAuthFunction);
+    }
+
     /**
      * List partitions for table.
      *

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

@@ -213,6 +213,18 @@ public String partitions(String databaseName, String objectName) {
                 PARTITIONS);
     }
 
+    public String dropPartitions(String databaseName, String objectName) {
+        return SLASH.join(
+                V1,
+                prefix,
+                DATABASES,
+                encodeString(databaseName),
+                TABLES,
+                encodeString(objectName),
+                PARTITIONS,
+                "drop");
+    }
+
     public String markDonePartitions(String databaseName, String objectName) {
         return SLASH.join(
                 V1,

paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java

@@ -0,0 +1,67 @@
+/*
+ * 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.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 javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+
+/** Request for creating partitions. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CreatePartitionsRequest implements RESTRequest {
+
+    private static final String FIELD_PARTITION_SPECS = "partitionSpecs";
+    private static final String FIELD_IGNORE_IF_EXISTS = "ignoreIfExists";
+
+    @JsonProperty(FIELD_PARTITION_SPECS)
+    private final List<Map<String, String>> partitionSpecs;
+
+    @JsonProperty(FIELD_IGNORE_IF_EXISTS)
+    private final boolean ignoreIfExists;
+
+    public CreatePartitionsRequest(List<Map<String, String>> partitionSpecs) {
+        this(partitionSpecs, true);
+    }
+
+    @JsonCreator
+    public CreatePartitionsRequest(
+            @JsonProperty(FIELD_PARTITION_SPECS) List<Map<String, String>> partitionSpecs,
+            @JsonProperty(FIELD_IGNORE_IF_EXISTS) @Nullable Boolean ignoreIfExists) {
+        this.partitionSpecs = partitionSpecs;
+        this.ignoreIfExists = ignoreIfExists == null || ignoreIfExists;
+    }
+
+    @JsonGetter(FIELD_PARTITION_SPECS)
+    public List<Map<String, String>> getPartitionSpecs() {
+        return partitionSpecs;
+    }
+
+    @JsonGetter(FIELD_IGNORE_IF_EXISTS)
+    public boolean ignoreIfExists() {
+        return ignoreIfExists;
+    }
+}

paimon-api/src/main/java/org/apache/paimon/rest/requests/DropPartitionsRequest.java

@@ -0,0 +1,67 @@
+/*
+ * 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.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 javax.annotation.Nullable;
+
+import java.util.List;
+import java.util.Map;
+
+/** Request for dropping (unregistering) partitions. */
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DropPartitionsRequest implements RESTRequest {
+
+    private static final String FIELD_PARTITION_SPECS = "partitionSpecs";
+    private static final String FIELD_IGNORE_IF_NOT_EXISTS = "ignoreIfNotExists";
+
+    @JsonProperty(FIELD_PARTITION_SPECS)
+    private final List<Map<String, String>> partitionSpecs;
+
+    @JsonProperty(FIELD_IGNORE_IF_NOT_EXISTS)
+    private final boolean ignoreIfNotExists;
+
+    public DropPartitionsRequest(List<Map<String, String>> partitionSpecs) {
+        this(partitionSpecs, true);
+    }
+
+    @JsonCreator
+    public DropPartitionsRequest(
+            @JsonProperty(FIELD_PARTITION_SPECS) List<Map<String, String>> partitionSpecs,
+            @JsonProperty(FIELD_IGNORE_IF_NOT_EXISTS) @Nullable Boolean ignoreIfNotExists) {
+        this.partitionSpecs = partitionSpecs;
+        this.ignoreIfNotExists = ignoreIfNotExists == null || ignoreIfNotExists;
+    }
+
+    @JsonGetter(FIELD_PARTITION_SPECS)
+    public List<Map<String, String>> getPartitionSpecs() {
+        return partitionSpecs;
+    }
+
+    @JsonGetter(FIELD_IGNORE_IF_NOT_EXISTS)
+    public boolean ignoreIfNotExists() {
+        return ignoreIfNotExists;
+    }
+}