apache · sundapeng · Sep 1, 2026 · Sep 1, 2026 · Sep 1, 2026 · Sep 1, 2026
diff --git a/docs/docs/concepts/rest/rest-api.md b/docs/docs/concepts/rest/rest-api.md
@@ -25,6 +25,10 @@ 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.
 
+Custom partition locations extend the existing `POST .../partitions` request. When present,
+`partitionLocations` has the same length and order as `partitionSpecs`; a null entry uses the
+default location. Servers that do not support custom partition locations reject the request.
+
 <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 cannot register a custom partition `LOCATION` through SQL. An upgraded Flink reader can read
+one registered through Spark or the catalog API; upgrade every reader before registering it.
+
 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,16 @@ 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 its data. It requires
+a supporting REST catalog server. The location must be outside the table directory and must not
+overlap another partition location. Default and custom locations can coexist in one table.
+
+Custom-location partitions use the client's filesystem credentials and are read-only. `DROP
+PARTITION` only unregisters them, `MSCK REPAIR TABLE` preserves them, and `ANALYZE TABLE` rejects
+them.
+
+Upgrade the REST server and every reader before adding custom partition locations.
+
 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,47 @@ 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('createPartitionsWithLocations'),
+    'custom locations 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 location responses',
+  );
+  const createRequestProperties = contract.requireProperties('CreatePartitionsRequest', [
+    'partitionSpecs',
+    'partitionLocations',
+  ]);
+  const requestLocations = createRequestProperties.partitionLocations;
+  contract.checkSpec(
+    Array.isArray(requestLocations.type) &&
+      requestLocations.type.includes('array') &&
+      requestLocations.type.includes('null') &&
+      Array.isArray(requestLocations.items.type) &&
+      requestLocations.items.type.includes('string') &&
+      requestLocations.items.type.includes('null'),
+    'CreatePartitionsRequest.partitionLocations must be a nullable array of nullable strings',
+  );
+  contract.checkSpec(
+    requestLocations.description.includes('partitionSpecs') &&
+      requestLocations.description.toLowerCase().includes('position') &&
+      requestLocations.description.toLowerCase().includes('same length'),
+    'CreatePartitionsRequest.partitionLocations must document positional alignment and equal length',
+  );
   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,11 @@ paths:
       tags:
         - partition
       summary: Create partitions
+      description: >
+        Creates partitions, optionally with custom locations aligned to partitionSpecs by position.
+        A server that receives partitionLocations must validate and persist the complete partition,
+        location, and statistics change atomically. A request without partitionLocations retains
+        the legacy behavior and uses derived default locations.
       operationId: createPartitions
       parameters:
         - name: prefix
@@ -1014,6 +1019,8 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/CreatePartitionsResponse'
+        "400":
+          $ref: '#/components/responses/BadRequestErrorResponse'
         "401":
           $ref: '#/components/responses/UnauthorizedErrorResponse'
         "404":
@@ -1022,6 +1029,12 @@ paths:
           $ref: '#/components/responses/ResourceAlreadyExistErrorResponse'
         "500":
           $ref: '#/components/responses/ServerErrorResponse'
+        "501":
+          description: The catalog provider does not support custom partition locations
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ErrorResponse'
   /v1/{prefix}/databases/{database}/tables/{table}/partitions/drop:
     post:
       tags:
@@ -2497,6 +2510,11 @@ 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" ]
+        partitionLocations:
+          description: Optional custom locations aligned with partitionSpecs by position. The list must have the same length as partitionSpecs; a null item uses the default location. The server must reject the whole request if it cannot atomically validate and store every location.
+          type: [ array, "null" ]
+          items:
+            type: [ string, "null" ]
     CreatePartitionsResponse:
       type: object
       required:
@@ -3766,6 +3784,9 @@ components:
           type: object
           additionalProperties:
             type: string
+        location:
+          description: Custom absolute location of this partition. When absent, the location is derived from the table root and partition spec.
+          type: string
     PartitionStatistics:
       type: object
       properties:

diff --git a/paimon-api/src/main/java/org/apache/paimon/partition/Partition.java b/paimon-api/src/main/java/org/apache/paimon/partition/Partition.java
@@ -36,7 +36,7 @@
 import static org.apache.paimon.rest.responses.AuditRESTResponse.FIELD_UPDATED_AT;
 import static org.apache.paimon.rest.responses.AuditRESTResponse.FIELD_UPDATED_BY;
 
-/** Represent a partition, including statistics and done flag. */
+/** Represents a partition, including statistics, completion state, options, and location. */
 @JsonIgnoreProperties(ignoreUnknown = true)
 @Public
 public class Partition extends PartitionStatistics {
@@ -45,6 +45,7 @@ public class Partition extends PartitionStatistics {
 
     public static final String FIELD_DONE = "done";
     public static final String FIELD_OPTIONS = "options";
+    public static final String FIELD_LOCATION = "location";
 
     @JsonProperty(FIELD_DONE)
     private final boolean done;
@@ -74,6 +75,11 @@ public class Partition extends PartitionStatistics {
     @Nullable
     private final Map<String, String> options;
 
+    @JsonProperty(FIELD_LOCATION)
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Nullable
+    private final String location;
+
     public Partition(
             Map<String, String> spec,
             long recordCount,
@@ -87,13 +93,44 @@ public Partition(
             @Nullable Long updatedAt,
             @Nullable String updatedBy,
             @Nullable Map<String, String> options) {
+        this(
+                spec,
+                recordCount,
+                fileSizeInBytes,
+                fileCount,
+                lastFileCreationTime,
+                totalBuckets,
+                done,
+                createdAt,
+                createdBy,
+                updatedAt,
+                updatedBy,
+                options,
+                null);
+    }
+
+    public Partition(
+            Map<String, String> spec,
+            long recordCount,
+            long fileSizeInBytes,
+            long fileCount,
+            long lastFileCreationTime,
+            int totalBuckets,
+            boolean done,
+            @Nullable Long createdAt,
+            @Nullable String createdBy,
+            @Nullable Long updatedAt,
+            @Nullable String updatedBy,
+            @Nullable Map<String, String> options,
+            @Nullable String location) {
         super(spec, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime, totalBuckets);
         this.done = done;
         this.createdAt = createdAt;
         this.createdBy = createdBy;
         this.updatedAt = updatedAt;
         this.updatedBy = updatedBy;
         this.options = options;
+        this.location = location;
     }
 
     public Partition(
@@ -137,7 +174,8 @@ static Partition fromJson(
             @JsonProperty(FIELD_CREATED_BY) @Nullable String createdBy,
             @JsonProperty(FIELD_UPDATED_AT) @Nullable Long updatedAt,
             @JsonProperty(FIELD_UPDATED_BY) @Nullable String updatedBy,
-            @JsonProperty(FIELD_OPTIONS) @Nullable Map<String, String> options) {
+            @JsonProperty(FIELD_OPTIONS) @Nullable Map<String, String> options,
+            @JsonProperty(FIELD_LOCATION) @Nullable String location) {
         return new Partition(
                 spec,
                 orUnknown(recordCount),
@@ -150,7 +188,8 @@ static Partition fromJson(
                 createdBy,
                 updatedAt,
                 updatedBy,
-                options);
+                options,
+                location);
     }
 
     private static long orUnknown(@Nullable Long value) {
@@ -192,6 +231,13 @@ public Map<String, String> options() {
         return options;
     }
 
+    /** Custom partition location, or null when it is derived from the table root and spec. */
+    @Nullable
+    @JsonGetter(FIELD_LOCATION)
+    public String location() {
+        return location;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) {
@@ -206,13 +252,21 @@ public boolean equals(Object o) {
                 && Objects.equals(createdBy, partition.createdBy)
                 && Objects.equals(updatedAt, partition.updatedAt)
                 && Objects.equals(updatedBy, partition.updatedBy)
-                && Objects.equals(options, partition.options);
+                && Objects.equals(options, partition.options)
+                && Objects.equals(location, partition.location);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(
-                super.hashCode(), done, createdAt, createdBy, updatedAt, updatedBy, options);
+                super.hashCode(),
+                done,
+                createdAt,
+                createdBy,
+                updatedAt,
+                updatedBy,
+                options,
+                location);
     }
 
     @Override
@@ -242,6 +296,8 @@ public String toString() {
                 + updatedBy
                 + ", options="
                 + options
+                + ", location="
+                + location
                 + '}';
     }
 }
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,24 @@ 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 partitionLocations custom locations aligned with {@code partitions} by position, where
+     *     null entries use derived defaults; null omits the location extension
      * @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<String> partitionLocations) {
         CreatePartitionsRequest request =
                 new CreatePartitionsRequest(
                         partitions,
                         ignoreIfExists,
                         statistics,
-                        statistics == null ? null : replaceStatistics);
+                        statistics == null ? null : replaceStatistics,
+                        partitionLocations);
         return client.post(
                 resourcePaths.partitions(identifier.getDatabaseName(), identifier.getObjectName()),
                 request,