apache · JingsongLi · Sep 3, 2026 · Sep 3, 2026 · Sep 3, 2026 · Sep 3, 2026
diff --git a/docs/docs/concepts/rest/rest-api.md b/docs/docs/concepts/rest/rest-api.md
@@ -25,6 +25,11 @@ under the License.
 The OpenAPI 3.1 document below defines the language-neutral wire contract for REST Catalog
 servers and clients. It can also be used to generate or validate SDK models in other languages.
 
+Partition options use the existing `POST .../partitions` request. `partitionOptions` follows the
+order of `partitionSpecs`; use `{}` when a partition has no options. Custom locations use the
+`path` option. Before registering custom locations, ensure that the REST server supports partition
+options and all readers support custom locations.
+
 <body>
     <iframe src="/docs/master/rest-catalog-open-api.yaml" width="100%" height="800px" />
 </body>
diff --git a/docs/docs/flink/sql-ddl.md b/docs/docs/flink/sql-ddl.md
@@ -49,6 +49,9 @@ and a table whose catalog holds no partitions reads as empty. Flink has no SQL c
 them: use Spark's `MSCK REPAIR TABLE` or the catalog's partition API. Flink writes on a current
 version do register the partitions they produce.
 
+Flink SQL cannot set a custom partition `LOCATION`. Upgrade Flink readers before registering custom
+locations through the catalog API.
+
 In a REST catalog, asking for catalog-managed partitions on a table that cannot have them — an
 external table, or `format-table.implementation = engine` — fails. In any other catalog the option
 keeps the meaning it has always had on a Format Table — none — and partitions come from the

diff --git a/docs/docs/spark/sql-ddl.md b/docs/docs/spark/sql-ddl.md
@@ -213,6 +213,8 @@ partitions and Spark supports the standard partition DDL:
 
 ```sql
 ALTER TABLE my_table ADD PARTITION (dt='2025-01-01');
+ALTER TABLE my_table ADD PARTITION (dt='2024-12-31')
+    LOCATION 'oss://archive-bucket/events/dt=2024-12-31';
 ALTER TABLE my_table DROP PARTITION (dt='2025-01-01');
 MSCK REPAIR TABLE my_table;
 SHOW PARTITIONS my_table;
@@ -226,6 +228,10 @@ On a Format Table whose partitions are discovered from the filesystem, `ADD PART
 added partition before any data is written returns no rows. `DROP PARTITION` unregisters the
 partition and deletes its directory.
 
+`ADD PARTITION ... LOCATION` registers a custom absolute URI without moving data and requires a
+compatible REST catalog. Paimon can read the partition but does not write, delete, or analyze its
+data. `DROP PARTITION` only unregisters it, and `MSCK REPAIR TABLE` leaves it unchanged.
+
 A partition value that is empty or all whitespace is rejected by `ADD PARTITION`, `DROP PARTITION`
 and `TRUNCATE PARTITION`. Such a value is written to the partition named by
 `partition.default-name` (`__DEFAULT_PARTITION__` unless configured otherwise), the same partition

diff --git a/docs/scripts/validate-rest-openapi.js b/docs/scripts/validate-rest-openapi.js
@@ -237,6 +237,48 @@ function validateCatalogOpenApi() {
     'dropTable',
   ].forEach(contract.requireOperation);
 
+  const createOperation = contract.requireOperation('createPartitions');
+  const createRequestRef =
+    createOperation.requestBody.content['application/json'].schema.$ref;
+  contract.checkSpec(
+    createRequestRef === '#/components/schemas/CreatePartitionsRequest',
+    'createPartitions must use CreatePartitionsRequest',
+  );
+  contract.checkSpec(
+    !contract.operations.has('createPartitionsWithOptions'),
+    'partition options must use the existing createPartitions operation',
+  );
+  const createResponseRef =
+    createOperation.responses['200'].content['application/json'].schema.$ref;
+  contract.checkSpec(
+    createResponseRef === '#/components/schemas/CreatePartitionsResponse',
+    'createPartitions must use CreatePartitionsResponse',
+  );
+  contract.checkSpec(
+    createOperation.responses['400'] && createOperation.responses['501'],
+    'createPartitions must declare invalid and unsupported option responses',
+  );
+  const createRequestProperties = contract.requireProperties('CreatePartitionsRequest', [
+    'partitionSpecs',
+    'partitionOptions',
+  ]);
+  const requestOptions = createRequestProperties.partitionOptions;
+  contract.checkSpec(
+    Array.isArray(requestOptions.type) &&
+      requestOptions.type.includes('array') &&
+      requestOptions.type.includes('null') &&
+      requestOptions.items.type === 'object' &&
+      requestOptions.items.additionalProperties.type === 'string',
+    'CreatePartitionsRequest.partitionOptions must be a nullable array of string maps',
+  );
+  contract.checkSpec(
+    requestOptions.description.includes('partitionSpecs') &&
+      requestOptions.description.toLowerCase().includes('position') &&
+      requestOptions.description.toLowerCase().includes('same length') &&
+      requestOptions.description.toLowerCase().includes('empty object'),
+    'CreatePartitionsRequest.partitionOptions must document positional alignment and empty options',
+  );
+  contract.requireProperties('Partition', ['options']);
   contract.requireProperties('ConfigResponse', ['defaults', 'overrides']);
   contract.requireProperties('CreateDatabaseRequest', ['name', 'options']);
   contract.requireProperties('AlterDatabaseRequest', ['removals', 'updates']);

diff --git a/docs/static/rest-catalog-open-api.yaml b/docs/static/rest-catalog-open-api.yaml
@@ -984,6 +984,9 @@ paths:
       tags:
         - partition
       summary: Create partitions
+      description: >
+        Creates partitions with optional position-aligned options. The server validates and stores
+        the partitions, options, and statistics atomically.
       operationId: createPartitions
       parameters:
         - name: prefix
@@ -1014,6 +1017,8 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/CreatePartitionsResponse'
+        "400":
+          $ref: '#/components/responses/BadRequestErrorResponse'
         "401":
           $ref: '#/components/responses/UnauthorizedErrorResponse'
         "404":
@@ -1022,6 +1027,12 @@ paths:
           $ref: '#/components/responses/ResourceAlreadyExistErrorResponse'
         "500":
           $ref: '#/components/responses/ServerErrorResponse'
+        "501":
+          description: The catalog provider does not support partition options
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
   /v1/{prefix}/databases/{database}/tables/{table}/partitions/drop:
     post:
       tags:
@@ -2497,6 +2508,13 @@ components:
         replaceStatistics:
           description: Whether partitionStatistics replace the stored values rather than add to them; required whenever partitionStatistics is present, and absent otherwise. Replacing overwrites recordCount, fileSizeInBytes, fileCount and lastFileCreationTime; adding sums the three counts and keeps the later lastFileCreationTime, since two timestamps do not add. A field reported as unknown leaves the stored one alone either way, and totalBuckets is never combined. A client that reports only the files it just wrote adds; one that reports a whole partition, such as an overwrite or a directory rescan, replaces.
           type: [ boolean, "null" ]
+        partitionOptions:
+          description: Optional options aligned with partitionSpecs by position. The list must have the same length as partitionSpecs; use an empty object when a partition has no options. A custom location is stored under the path key.
+          type: [ array, "null" ]
+          items:
+            type: object
+            additionalProperties:
+              type: string
     CreatePartitionsResponse:
       type: object
       required:

diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -996,20 +996,23 @@ public void markDonePartitions(Identifier identifier, List<Map<String, String>>
      *     PartitionStatistics#spec()} rather than by position, or null to report none
      * @param replaceStatistics whether the report replaces the stored values rather than adding to
      *     them; ignored when {@code statistics} is null, and not sent at all in that case
+     * @param partitionOptions options aligned with {@code partitions} by position, or null
      * @return the partitions the server created and the ones it already held
      */
     public CreatePartitionsResponse createPartitions(
             Identifier identifier,
             List<Map<String, String>> partitions,
             boolean ignoreIfExists,
             @Nullable List<PartitionStatistics> statistics,
-            boolean replaceStatistics) {
+            boolean replaceStatistics,
+            @Nullable List<Map<String, String>> partitionOptions) {
         CreatePartitionsRequest request =
                 new CreatePartitionsRequest(
                         partitions,
                         ignoreIfExists,
                         statistics,
-                        statistics == null ? null : replaceStatistics);
+                        statistics == null ? null : replaceStatistics,
+                        partitionOptions);
         return client.post(
                 resourcePaths.partitions(identifier.getDatabaseName(), identifier.getObjectName()),
                 request,

diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java b/paimon-api/src/main/java/org/apache/paimon/rest/requests/CreatePartitionsRequest.java
@@ -33,6 +33,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
 /**
  * Request for creating partitions.
  *
@@ -47,6 +49,7 @@ public class CreatePartitionsRequest implements RESTRequest {
     private static final String FIELD_IGNORE_IF_EXISTS = "ignoreIfExists";
     private static final String FIELD_PARTITION_STATISTICS = "partitionStatistics";
     private static final String FIELD_REPLACE_STATISTICS = "replaceStatistics";
+    private static final String FIELD_PARTITION_OPTIONS = "partitionOptions";
 
     @JsonProperty(FIELD_PARTITION_SPECS)
     private final List<Map<String, String>> partitionSpecs;
@@ -64,13 +67,26 @@ public class CreatePartitionsRequest implements RESTRequest {
     @Nullable
     private final Boolean replaceStatistics;
 
+    @JsonProperty(FIELD_PARTITION_OPTIONS)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Nullable
+    private final List<Map<String, String>> partitionOptions;
+
     public CreatePartitionsRequest(List<Map<String, String>> partitionSpecs) {
         this(partitionSpecs, true);
     }
 
     public CreatePartitionsRequest(
             List<Map<String, String>> partitionSpecs, @Nullable Boolean ignoreIfExists) {
-        this(partitionSpecs, ignoreIfExists, null, null);
+        this(partitionSpecs, ignoreIfExists, null, null, null);
+    }
+
+    public CreatePartitionsRequest(
+            List<Map<String, String>> partitionSpecs,
+            @Nullable Boolean ignoreIfExists,
+            @Nullable List<PartitionStatistics> partitionStatistics,
+            @Nullable Boolean replaceStatistics) {
+        this(partitionSpecs, ignoreIfExists, partitionStatistics, replaceStatistics, null);
     }
 
     @JsonCreator
@@ -79,11 +95,30 @@ public CreatePartitionsRequest(
             @JsonProperty(FIELD_IGNORE_IF_EXISTS) @Nullable Boolean ignoreIfExists,
             @JsonProperty(FIELD_PARTITION_STATISTICS) @Nullable
                     List<PartitionStatistics> partitionStatistics,
-            @JsonProperty(FIELD_REPLACE_STATISTICS) @Nullable Boolean replaceStatistics) {
+            @JsonProperty(FIELD_REPLACE_STATISTICS) @Nullable Boolean replaceStatistics,
+            @JsonProperty(FIELD_PARTITION_OPTIONS) @Nullable
+                    List<Map<String, String>> partitionOptions) {
+        checkArgument(
+                partitionOptions == null
+                        || (partitionSpecs != null
+                                && partitionOptions.size() == partitionSpecs.size()),
+                "partitionOptions must be null or have the same size as partitionSpecs.");
+        checkArgument(
+                partitionOptions == null || !partitionOptions.contains(null),
+                "partitionOptions must not contain null maps.");
+        checkArgument(
+                partitionOptions == null
+                        || partitionOptions.stream()
+                                .flatMap(options -> options.entrySet().stream())
+                                .noneMatch(
+                                        entry ->
+                                                entry.getKey() == null || entry.getValue() == null),
+                "partitionOptions must not contain null keys or values.");
         this.partitionSpecs = partitionSpecs;
         this.ignoreIfExists = ignoreIfExists == null || ignoreIfExists;
         this.partitionStatistics = partitionStatistics;
         this.replaceStatistics = replaceStatistics;
+        this.partitionOptions = partitionOptions;
     }
 
     @JsonGetter(FIELD_PARTITION_SPECS)
@@ -113,6 +148,13 @@ public Boolean replaceStatistics() {
         return replaceStatistics;
     }
 
+    /** Options aligned with partition specs; a null list omits the field. */
+    @JsonGetter(FIELD_PARTITION_OPTIONS)
+    @Nullable
+    public List<Map<String, String>> getPartitionOptions() {
+        return partitionOptions;
+    }
+
     /**
      * Registering is an upsert and replacing lands on the same value twice, so both survive being
      * sent again. Adding does not: a second delivery is counted again. A request that reports no

diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java
@@ -350,10 +350,16 @@ public void createPartitions(
             List<Map<String, String>> partitions,
             boolean ignoreIfExists,
             @Nullable List<PartitionStatistics> statistics,
-            boolean replaceStatistics)
+            boolean replaceStatistics,
+            @Nullable List<Map<String, String>> partitionOptions)
             throws TableNotExistException {
         wrapped.createPartitions(
-                identifier, partitions, ignoreIfExists, statistics, replaceStatistics);
+                identifier,
+                partitions,
+                ignoreIfExists,
+                statistics,
+                replaceStatistics,
+                partitionOptions);
         if (partitionCache != null) {
             partitionCache.invalidate(identifier);
         }

diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -1072,35 +1072,29 @@ default void createPartitions(Identifier identifier, List<Map<String, String>> p
             throws TableNotExistException {}
 
     /**
-     * Create partitions of the specify table, with explicit existence semantics and optionally
-     * reporting statistics for them in the same call.
-     *
-     * <p>The statistics are matched to {@code partitions} by {@link PartitionStatistics#spec()}, so
-     * they may cover only some of them, and {@code replaceStatistics} says whether they replace
-     * what the catalog already holds or add to it. What decides whether they survive is whether a
-     * catalog overrides this method: one that does not registers the partitions exactly as {@link
-     * #createPartitions(Identifier, List)} does and drops the report, however much of it the
-     * catalog could have stored, and for a catalog that keeps no partitions at all that means it
-     * does nothing.
-     *
-     * @param identifier path of the table to create partitions
-     * @param partitions partitions to be created
-     * @param ignoreIfExists if false, fail when any partition already exists and apply none of the
-     *     batch; if true, behave like {@link #createPartitions(Identifier, List)}
-     * @param statistics statistics to report, or null to report none
-     * @param replaceStatistics whether the report replaces the stored values rather than adding to
-     *     them; ignored when {@code statistics} is null
-     * @throws TableNotExistException if the table does not exist
-     * @throws UnsupportedOperationException if {@code ignoreIfExists} is false and the catalog does
-     *     not implement strict creation, which is what the default here does
+     * Create partitions atomically unless existing entries are ignored, with optional statistics
+     * and position-aligned options.
      */
     default void createPartitions(
             Identifier identifier,
             List<Map<String, String>> partitions,
             boolean ignoreIfExists,
             @Nullable List<PartitionStatistics> statistics,
-            boolean replaceStatistics)
+            boolean replaceStatistics,
+            @Nullable List<Map<String, String>> partitionOptions)
             throws TableNotExistException {
+        if (partitionOptions != null) {
+            if (partitionOptions.size() != partitions.size() || partitionOptions.contains(null)) {
+                throw new IllegalArgumentException(
+                        "Partition options must contain one non-null map per partition.");
+            }
+            if (partitionOptions.stream().anyMatch(options -> !options.isEmpty())) {
+                throw new UnsupportedOperationException(
+                        String.format(
+                                "Catalog %s does not support partition options.",
+                                getClass().getName()));
+            }
+        }
         if (!ignoreIfExists) {
             throw new UnsupportedOperationException(
                     String.format(

diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -331,10 +331,16 @@ public void createPartitions(
             List<Map<String, String>> partitions,
             boolean ignoreIfExists,
             @Nullable List<PartitionStatistics> statistics,
-            boolean replaceStatistics)
+            boolean replaceStatistics,
+            @Nullable List<Map<String, String>> partitionOptions)
             throws TableNotExistException {
         wrapped.createPartitions(
-                identifier, partitions, ignoreIfExists, statistics, replaceStatistics);
+                identifier,
+                partitions,
+                ignoreIfExists,
+                statistics,
+                replaceStatistics,
+                partitionOptions);
     }
 
     @Override