You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "sundapeng (via GitHub)" <gi...@apache.org> on 2026/09/02 03:46:22 UTC
[PR] [core][rest][spark] Add explicit Format Table partition locations [paimon]
sundapeng opened a new pull request, #9540:
URL: https://github.com/apache/paimon/pull/9540
### Purpose
As a follow-up to #8750 and #8751, this adds Spark's standard explicit-location
form for registering existing data in a catalog-managed Format Table:
```sql
ALTER TABLE t ADD PARTITION (dt = '2026-09-02')
LOCATION 's3://bucket/existing/dt=2026-09-02';
```
For `ADD PARTITION ... LOCATION`, Spark canonicalizes and validates the absolute
URI before registration. It rejects roots, table ancestors, and the partition's
derived default directory, registers the location as partition metadata without
creating or probing the supplied directory, and verifies that the REST server
stored and echoed the same canonical value. Partitions added without `LOCATION`
keep their current derived-path behavior.
When explicit locations may exist, reads and mutation preflights validate the
complete catalog registry as one path-ownership set. Duplicate, ancestor,
descendant, table-root, and derived-path conflicts fail closed, as does a mismatch
between the authoritative explicit-location count and the listed registry.
Locations outside the table root use a `FileIO` resolved from the client's
`CatalogContext`; locations under the table root continue using the table's
`FileIO`.
Data in an explicitly located partition is read-only to Paimon. Write commits,
overwrites, truncates, and `ANALYZE TABLE` operations whose scope includes such a
partition fail before publishing or deleting files or mutating catalog metadata.
`DROP PARTITION` only unregisters it and never probes or deletes the explicit URI.
`MSCK REPAIR TABLE` leaves the registration intact and does not inspect or collect
statistics from that location.
The REST contract uses a dedicated `partitions/with-locations` resource and the
`format-table-partition-location-v1` capability token. A location-aware POST is
refused unless the server advertises the capability. Spark REST requests advertise
the client capability; Flink requests intentionally do not. The existing partition
resource and location-free payloads remain compatible.
This change is limited to catalog-managed Format Tables; it does not add
multi-location storage to native Paimon tables or change the table-level
`SET LOCATION` rejection from #9435.
### Tests
- 489 tests in 12 focused `paimon-api` and `paimon-core` suites covering JSON and
REST compatibility, catalog delegation and caching, partition management, scans,
reads, commits, `FileIO` routing, registry consistency, and path ownership.
- Spark 3: 96 focused Scala tests, 19 `FormatTablePartitionRepairTest` tests, and
2 REST capability-header tests.
- Spark 4: 83 focused Scala tests and 2 REST capability-header tests.
- Flink REST compatibility integration tests: 2.
- Checkstyle, Spotless, Maven Enforcer, and RAT.
- JDK 8 `clean package`.
- REST OpenAPI validation: 57 Catalog operations and 6 Management operations.
- Docusaurus production build.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][rest][spark] Add custom Format Table partition locations [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9540:
URL: https://github.com/apache/paimon/pull/9540#discussion_r3921081037
##########
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);
Review Comment:
**[P1] Do not send locations through a route that legacy servers accept silently.** A pre-feature server deserializes this request with `@JsonIgnoreProperties(ignoreUnknown = true)`, so it discards `partitionLocations`, registers the spec at the derived default location, and returns the unchanged successful response. The new client therefore treats `ADD PARTITION ... LOCATION` as successful even though later scans will not read the requested external directory. The current mock `501` test only models an upgraded server that recognizes the field. Please use a location-specific endpoint or a mandatory capability/response acknowledgement that an old server cannot accidentally satisfy, and add a mixed-version test against the base DTO behavior.
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java:
##########
@@ -203,6 +207,8 @@ public void commit(List<CommitMessage> commitMessages) {
}
}
+ List<Partition> validatedPartitions = rejectWritesToCustomLocationPartitions(messages);
Review Comment:
**[P1] Make the default-location ownership check atomic with registration.** This is only a preflight: after it returns, a concurrent `ADD PARTITION ... LOCATION` can register one of the affected specs before the writer publishes and calls `createPartitions` with no location assertion. Because `ignoreIfExists` is true, registration succeeds against the now-custom partition and its statistics may be attached to that partition even though the new files were published under the default path, making the committed rows invisible. Please pass a position-aligned all-null `partitionLocations` list for these preflighted default-owned specs so the server rejects an intervening custom owner (and abort removes the published files), and cover the interleaving with a latch-based test.
##########
paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java:
##########
@@ -1072,35 +1072,23 @@ 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 locations whose null entries use defaults.
*/
default void createPartitions(
Identifier identifier,
List<Map<String, String>> partitions,
boolean ignoreIfExists,
@Nullable List<PartitionStatistics> statistics,
- boolean replaceStatistics)
+ boolean replaceStatistics,
+ @Nullable List<String> partitionLocations)
Review Comment:
**[P1] Preserve the existing public overload.** This replaces the five-argument `createPartitions(identifier, partitions, ignoreIfExists, statistics, replaceStatistics)` method that exists on the base branch of the `@Public` `Catalog` interface. Existing source integrations will no longer compile and already-compiled callers can fail with `NoSuchMethodError`; an existing catalog implementation that overrides only the old method is also bypassed by new six-argument callers, potentially dropping statistics through the default fallback. Please keep the old overload and make the new overload delegate to it when `partitionLocations` is null (and retain analogous bridges in `RESTApi` and `FormatTablePartitionManager`).
##########
paimon-python/pypaimon/table/format/format_table.py:
##########
@@ -89,6 +92,44 @@ def format(self) -> Format:
def options(self) -> Dict[str, str]:
return self._options
+ def _has_custom_partition_locations(self) -> bool:
+ catalog_environment = getattr(self, "_catalog_environment", None)
+ if catalog_environment is None:
+ return False
+ catalog = catalog_environment.catalog_loader.load()
+ page_token = None
+ seen_page_tokens = set()
+ while True:
+ page = catalog.list_partitions_paged(
Review Comment:
**[P2] Avoid scanning the complete registry before every normal read.** When no custom location exists, this loop must drain every page, and `FormatTableScan.plan()` invokes it unconditionally for every catalog-managed Format Table read (the first write does the same). A table with one million default-location partitions adds roughly 1,000 serial REST requests before planning can begin. Please expose an authoritative server-side has/count signal or negotiate a capability that lets PyPaimon reject unsupported custom-location tables without enumerating the entire registry.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][rest][spark] Add custom Format Table partition locations [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9540:
URL: https://github.com/apache/paimon/pull/9540#discussion_r3921081037
##########
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);
Review Comment:
**[P1] Do not send locations through a route that legacy servers accept silently.** A pre-feature server deserializes this request with `@JsonIgnoreProperties(ignoreUnknown = true)`, so it discards `partitionLocations`, registers the spec at the derived default location, and returns the unchanged successful response. The new client therefore treats `ADD PARTITION ... LOCATION` as successful even though later scans will not read the requested external directory. The current mock `501` test only models an upgraded server that recognizes the field. Please use a location-specific endpoint or a mandatory capability/response acknowledgement that an old server cannot accidentally satisfy, and add a mixed-version test against the base DTO behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [api][core][spark] Support custom partition locations [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #9540:
URL: https://github.com/apache/paimon/pull/9540
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][rest][spark] Add custom Format Table partition locations [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9540:
URL: https://github.com/apache/paimon/pull/9540#discussion_r3921799997
##########
paimon-api/src/main/java/org/apache/paimon/partition/Partition.java:
##########
@@ -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;
Review Comment:
More importantly, it is a highly specific variable—not a general-purpose one—so it should be tucked into the `options` object.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][rest][spark] Add custom Format Table partition locations [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9540:
URL: https://github.com/apache/paimon/pull/9540#discussion_r3921081043
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java:
##########
@@ -203,6 +207,8 @@ public void commit(List<CommitMessage> commitMessages) {
}
}
+ List<Partition> validatedPartitions = rejectWritesToCustomLocationPartitions(messages);
Review Comment:
**[P1] Make the default-location ownership check atomic with registration.** This is only a preflight: after it returns, a concurrent `ADD PARTITION ... LOCATION` can register one of the affected specs before the writer publishes and calls `createPartitions` with no location assertion. Because `ignoreIfExists` is true, registration succeeds against the now-custom partition and its statistics may be attached to that partition even though the new files were published under the default path, making the committed rows invisible. Please pass a position-aligned all-null `partitionLocations` list for these preflighted default-owned specs so the server rejects an intervening custom owner (and abort removes the published files), and cover the interleaving with a latch-based test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][rest][spark] Add custom Format Table partition locations [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9540:
URL: https://github.com/apache/paimon/pull/9540#discussion_r3921081034
##########
paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java:
##########
@@ -1072,35 +1072,23 @@ 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 locations whose null entries use defaults.
*/
default void createPartitions(
Identifier identifier,
List<Map<String, String>> partitions,
boolean ignoreIfExists,
@Nullable List<PartitionStatistics> statistics,
- boolean replaceStatistics)
+ boolean replaceStatistics,
+ @Nullable List<String> partitionLocations)
Review Comment:
**[P1] Preserve the existing public overload.** This replaces the five-argument `createPartitions(identifier, partitions, ignoreIfExists, statistics, replaceStatistics)` method that exists on the base branch of the `@Public` `Catalog` interface. Existing source integrations will no longer compile and already-compiled callers can fail with `NoSuchMethodError`; an existing catalog implementation that overrides only the old method is also bypassed by new six-argument callers, potentially dropping statistics through the default fallback. Please keep the old overload and make the new overload delegate to it when `partitionLocations` is null (and retain analogous bridges in `RESTApi` and `FormatTablePartitionManager`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][rest][spark] Add explicit Format Table partition locations [paimon]
Posted by "sundapeng (via GitHub)" <gi...@apache.org>.
sundapeng closed pull request #9540: [core][rest][spark] Add explicit Format Table partition locations
URL: https://github.com/apache/paimon/pull/9540
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][rest][spark] Add custom Format Table partition locations [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9540:
URL: https://github.com/apache/paimon/pull/9540#discussion_r3921299232
##########
paimon-api/src/main/java/org/apache/paimon/partition/Partition.java:
##########
@@ -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;
Review Comment:
I think location should be options, just like Table. Create partitions should also support options instead of location.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [core][rest][spark] Add explicit Format Table partition locations [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #9540:
URL: https://github.com/apache/paimon/pull/9540#discussion_r3914482020
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala:
##########
@@ -241,101 +266,194 @@ case class PaimonFormatTable(table: FormatTable)
rows: Array[InternalRow],
maps: Array[JMap[String, String]],
ignoreIfExists: Boolean): Unit = {
- if (maps.exists(_.keySet().asScala.exists(_.equalsIgnoreCase("location")))) {
- throw new UnsupportedOperationException(
- s"ADD PARTITION with LOCATION is not supported for Format Table ${table.fullName()}.")
- }
val onlyValueInPath =
CoreOptions.fromMap(table.options()).formatTablePartitionOnlyValueInPath()
val partitionKeys = table.partitionKeys().asScala.toSeq
rows.foreach(row => requireNameablePartitionValues("ADD PARTITION", row, partitionKeys))
- val specs = rows.map(row => toPaimonPartition(row, partitionKeys.take(row.numFields))).toSeq
- // Resolve (and path-safety validate) every directory before mutating anything.
+ val partitions = rows
+ .zip(maps)
+ .map {
+ case (row, properties) =>
+ val spec = toPaimonPartition(row, partitionKeys.take(row.numFields))
+ val location = properties.asScala.collectFirst {
+ case (key, value) if key.equalsIgnoreCase("location") =>
+ normalizeExplicitPartitionLocation(value, spec, onlyValueInPath)
+ }
+ spec -> location
+ }
+ .toSeq
+ val specs = partitions.map(_._1)
+ // Resolve (and path-safety validate) every default directory before mutating anything.
val partitionPaths =
- specs.map(spec => resolvePartitionPathWithinTable(orderedSpec(spec), onlyValueInPath))
- requirePartitionManager().createPartitions(specs.asJava, ignoreIfExists)
+ partitions.collect {
+ case (spec, None) =>
+ resolvePartitionPathWithinTable(orderedSpec(spec), onlyValueInPath)
+ }
+ val locations = partitions.collect {
+ case (spec, Some(location)) => new PartitionLocation(spec, location)
+ }
+ if (locations.isEmpty) {
+ requirePartitionManager().createPartitions(specs.asJava, ignoreIfExists)
+ } else {
+ requirePartitionManager()
+ .createPartitions(specs.asJava, ignoreIfExists, null, false, locations.asJava)
Review Comment:
**[P1] Validate the prospective registry atomically before registering locations.**
This path validates each explicit location only against the table root and the default directory of that same spec, then immediately mutates the catalog. A batch can therefore assign the same or nested `LOCATION` values to different specs; MSCK can similarly add a default spec whose directory is already owned by another explicit spec. Registration succeeds, but scans, writes, and DROP later reject the full registry as overlapping, so normal SQL cannot even unregister the bad entry. Please make the catalog/REST mutation atomically validate the complete post-update ownership set for both ordinary create and `/with-locations`; a client-side listing alone would still race concurrent ADDs.
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionPathResolver.java:
##########
@@ -0,0 +1,386 @@
+/*
+ * 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.table.format;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.PartitionPathUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayOutputStream;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+
+/** Resolves catalog-managed Format Table partition paths and rejects ambiguous catalog metadata. */
+public final class FormatTablePartitionPathResolver {
+
+ private final Path tablePath;
+ private final String tableName;
+ private final boolean onlyValueInPath;
+ private final Map<Map<String, String>, ResolvedPath> pathsBySpec = new LinkedHashMap<>();
+ private final Map<FileSystemKey, OwnershipNode> ownershipRoots = new HashMap<>();
+
+ FormatTablePartitionPathResolver(Path tablePath, String tableName, boolean onlyValueInPath) {
+ this.tablePath = tablePath;
+ this.tableName = tableName;
+ this.onlyValueInPath = onlyValueInPath;
+ }
+
+ Path resolve(LinkedHashMap<String, String> spec, @Nullable String explicitLocation) {
+ Path defaultPath =
+ new Path(
+ tablePath,
+ PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath));
+ if (explicitLocation == null) {
+ return defaultPath;
+ }
+
+ try {
+ return resolveExplicitLocation(tablePath, spec, onlyValueInPath, explicitLocation);
+ } catch (IllegalArgumentException e) {
+ throw invalidLocation(spec);
+ }
+ }
+
+ /**
+ * Canonicalizes an explicit partition location and verifies that it cannot name the table, its
+ * ancestor, or the partition's default directory.
+ */
+ public static Path resolveExplicitLocation(
+ Path tablePath,
+ LinkedHashMap<String, String> spec,
+ boolean onlyValueInPath,
+ String explicitLocation) {
+ PartitionPathUtils.validatePartitionSpecForPath(spec, onlyValueInPath);
+ Path defaultPath =
+ new Path(
+ tablePath,
+ PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath));
+ Path explicitPath = canonicalizeExplicitLocation(explicitLocation);
+ if (sameLocation(explicitPath, defaultPath)
+ || sameLocation(explicitPath, tablePath)
+ || isAncestor(explicitPath, tablePath)) {
+ throw new IllegalArgumentException("Explicit partition location overlaps table data.");
+ }
+ return explicitPath;
+ }
+
+ /**
+ * Records a resolved path. Returns false only for an identical duplicate entry for the same
+ * spec, which must not duplicate every row in that partition.
+ */
+ boolean validateAndRecord(LinkedHashMap<String, String> spec, Path path) {
+ ResolvedPath resolved = ResolvedPath.of(path);
+ ResolvedPath previousForSpec = pathsBySpec.get(spec);
+ if (previousForSpec != null) {
+ if (previousForSpec.equals(resolved)) {
+ return false;
+ }
+ throw overlappingLocations();
+ }
+
+ if (overlapsOwnedPath(resolved)) {
+ throw overlappingLocations();
+ }
+ pathsBySpec.put(new LinkedHashMap<>(spec), resolved);
+ return true;
+ }
+
+ private boolean overlapsOwnedPath(ResolvedPath path) {
+ OwnershipNode node =
+ ownershipRoots.computeIfAbsent(path.fileSystem(), ignored -> new OwnershipNode());
+ String[] segments = path.pathSegments();
+ for (String segment : segments) {
+ // A terminal node reached before the candidate ends is an existing ancestor.
+ if (node.owned) {
+ return true;
+ }
+ node = node.children.computeIfAbsent(segment, ignored -> new OwnershipNode());
+ }
+ // A terminal final node is equality. Children below it make the candidate an ancestor.
+ if (node.owned || !node.children.isEmpty()) {
+ return true;
+ }
+ node.owned = true;
+ return false;
+ }
+
+ /** Returns the canonical URI representation used on the partition-location wire contract. */
+ public static Path canonicalizeExplicitLocation(String location) {
+ String decoded = location;
+ try {
+ while (true) {
+ validateDecodedLocation(decoded);
+ String next = decodePercentOnce(decoded);
+ if (next.equals(decoded)) {
+ break;
+ }
+ decoded = next;
+ }
+
+ Path path = new Path(decoded);
+ URI uri = path.toUri();
+ String scheme = uri.getScheme();
+ String authority = uri.getAuthority();
+ String uriPath = uri.getPath();
+ if (scheme == null
+ || scheme.isEmpty()
+ || uri.getUserInfo() != null
+ || uriPath == null
+ || !uriPath.startsWith(Path.SEPARATOR)
+ || uriPath.equals(Path.SEPARATOR)) {
+ throw new IllegalArgumentException("Invalid explicit partition location.");
+ }
+
+ scheme = scheme.toLowerCase(Locale.ROOT);
+ if (!scheme.equals("file") && (authority == null || authority.isEmpty())) {
+ throw new IllegalArgumentException("Invalid explicit partition location.");
+ }
+ authority =
+ authority == null || authority.isEmpty()
+ ? null
+ : authority.toLowerCase(Locale.ROOT);
+ return new Path(scheme, authority, uriPath);
+ } catch (IllegalArgumentException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException("Invalid explicit partition location.", e);
+ }
+ }
+
+ private static void validateDecodedLocation(String location) {
+ if (location == null
+ || location.isEmpty()
+ || isBoundaryWhitespace(location)
+ || location.contains("?")
+ || location.contains("#")
+ || location.contains("\\")) {
+ throw new IllegalArgumentException("Invalid explicit partition location.");
+ }
+
+ for (int offset = 0; offset < location.length(); ) {
+ int codePoint = location.codePointAt(offset);
+ if (Character.isISOControl(codePoint)) {
+ throw new IllegalArgumentException("Invalid explicit partition location.");
+ }
+ offset += Character.charCount(codePoint);
+ }
+
+ for (String segment : location.split(Path.SEPARATOR, -1)) {
+ if (segment.equals(Path.CUR_DIR) || segment.equals("..")) {
+ throw new IllegalArgumentException("Invalid explicit partition location.");
+ }
+ }
+ }
+
+ private static boolean isBoundaryWhitespace(String value) {
+ int first = value.codePointAt(0);
+ int last = value.codePointBefore(value.length());
+ return isWhitespace(first) || isWhitespace(last);
+ }
+
+ private static boolean isWhitespace(int codePoint) {
+ return Character.isWhitespace(codePoint) || Character.isSpaceChar(codePoint);
+ }
+
+ private static String decodePercentOnce(String value) {
+ StringBuilder decoded = new StringBuilder(value.length());
+ for (int offset = 0; offset < value.length(); ) {
+ char current = value.charAt(offset);
+ if (current != '%') {
+ decoded.append(current);
+ offset++;
+ continue;
+ }
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ while (offset < value.length() && value.charAt(offset) == '%') {
+ if (offset + 2 >= value.length()) {
+ throw new IllegalArgumentException("Invalid percent encoding in location.");
+ }
+ int high = Character.digit(value.charAt(offset + 1), 16);
+ int low = Character.digit(value.charAt(offset + 2), 16);
+ if (high < 0 || low < 0) {
+ throw new IllegalArgumentException("Invalid percent encoding in location.");
+ }
+ bytes.write((high << 4) + low);
+ offset += 3;
+ }
+ try {
+ decoded.append(
+ StandardCharsets.UTF_8
+ .newDecoder()
+ .onMalformedInput(CodingErrorAction.REPORT)
+ .onUnmappableCharacter(CodingErrorAction.REPORT)
+ .decode(ByteBuffer.wrap(bytes.toByteArray())));
+ } catch (CharacterCodingException e) {
+ throw new IllegalArgumentException("Invalid percent encoding in location.", e);
+ }
+ }
+ return decoded.toString();
+ }
+
+ static boolean isWithin(Path candidate, Path root) {
+ ResolvedPath candidatePath = ResolvedPath.of(candidate);
+ ResolvedPath rootPath = ResolvedPath.of(root);
+ return rootPath.equals(candidatePath) || rootPath.isAncestorOf(candidatePath);
+ }
+
+ private static boolean sameLocation(Path left, Path right) {
+ return ResolvedPath.of(left).equals(ResolvedPath.of(right));
+ }
+
+ private static boolean isAncestor(Path candidateAncestor, Path candidateChild) {
+ return ResolvedPath.of(candidateAncestor).isAncestorOf(ResolvedPath.of(candidateChild));
+ }
+
+ private IllegalStateException invalidLocation(Map<String, String> spec) {
+ return new IllegalStateException(
+ String.format(
+ "Catalog returned an invalid explicit location for partition %s of Format Table %s.",
+ spec, tableName));
+ }
+
+ private IllegalStateException overlappingLocations() {
+ return new IllegalStateException(
+ String.format(
+ "Catalog returned overlapping locations for different partitions of Format Table %s.",
+ tableName));
+ }
+
+ /**
+ * One trie is maintained per filesystem. Visiting each path segment once is sufficient:
+ * ancestors are terminal nodes on the route, equality is the terminal node at the route's end,
+ * and descendants are children below that node.
+ */
+ private static final class OwnershipNode {
+
+ private final Map<String, OwnershipNode> children = new HashMap<>();
+ private boolean owned;
+ }
+
+ private static final class FileSystemKey {
+
+ private final String scheme;
+ private final String authority;
+
+ private FileSystemKey(String scheme, String authority) {
+ this.scheme = scheme;
+ this.authority = authority;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ FileSystemKey that = (FileSystemKey) o;
+ return scheme.equals(that.scheme) && authority.equals(that.authority);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(scheme, authority);
+ }
+ }
+
+ private static final class ResolvedPath {
+
+ private final String scheme;
+ private final String authority;
+ private final String path;
+
+ private ResolvedPath(String scheme, String authority, String path) {
+ this.scheme = scheme;
+ this.authority = authority;
+ this.path = path;
+ }
+
+ private static ResolvedPath of(Path path) {
+ URI uri = path.toUri().normalize();
+ String scheme = uri.getScheme();
+ // An absolute path without a scheme and file:/ name the same local filesystem.
+ scheme = scheme == null ? "file" : scheme.toLowerCase(Locale.ROOT);
+ String authority = uri.getAuthority();
+ authority = authority == null ? "" : authority.toLowerCase(Locale.ROOT);
Review Comment:
**[P1] Canonicalize the filesystem identity before comparing ownership.**
The ownership key uses the raw lower-cased scheme and authority, although different URI spellings can resolve to the same filesystem. For example, Hadoop can resolve `hdfs://nn` and `hdfs://nn:8020` to the same NameNode, but this code treats them as separate roots. A table at `hdfs://nn:8020/warehouse/t` can therefore accept an explicit location such as `hdfs://nn/warehouse/t`, bypassing the table-root check and potentially reading the whole table as one partition or writing through a default sibling into explicitly owned data. Please derive the key from the resolved/canonical FileSystem URI, including default ports and configured aliases, and use that identity consistently for overlap checks and FileIO routing.
##########
paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionPathResolver.java:
##########
@@ -0,0 +1,386 @@
+/*
+ * 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.table.format;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.PartitionPathUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayOutputStream;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+
+/** Resolves catalog-managed Format Table partition paths and rejects ambiguous catalog metadata. */
+public final class FormatTablePartitionPathResolver {
+
+ private final Path tablePath;
+ private final String tableName;
+ private final boolean onlyValueInPath;
+ private final Map<Map<String, String>, ResolvedPath> pathsBySpec = new LinkedHashMap<>();
+ private final Map<FileSystemKey, OwnershipNode> ownershipRoots = new HashMap<>();
+
+ FormatTablePartitionPathResolver(Path tablePath, String tableName, boolean onlyValueInPath) {
+ this.tablePath = tablePath;
+ this.tableName = tableName;
+ this.onlyValueInPath = onlyValueInPath;
+ }
+
+ Path resolve(LinkedHashMap<String, String> spec, @Nullable String explicitLocation) {
+ Path defaultPath =
+ new Path(
+ tablePath,
+ PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath));
+ if (explicitLocation == null) {
+ return defaultPath;
+ }
+
+ try {
+ return resolveExplicitLocation(tablePath, spec, onlyValueInPath, explicitLocation);
+ } catch (IllegalArgumentException e) {
+ throw invalidLocation(spec);
+ }
+ }
+
+ /**
+ * Canonicalizes an explicit partition location and verifies that it cannot name the table, its
+ * ancestor, or the partition's default directory.
+ */
+ public static Path resolveExplicitLocation(
+ Path tablePath,
+ LinkedHashMap<String, String> spec,
+ boolean onlyValueInPath,
+ String explicitLocation) {
+ PartitionPathUtils.validatePartitionSpecForPath(spec, onlyValueInPath);
+ Path defaultPath =
+ new Path(
+ tablePath,
+ PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath));
+ Path explicitPath = canonicalizeExplicitLocation(explicitLocation);
+ if (sameLocation(explicitPath, defaultPath)
+ || sameLocation(explicitPath, tablePath)
+ || isAncestor(explicitPath, tablePath)) {
+ throw new IllegalArgumentException("Explicit partition location overlaps table data.");
+ }
+ return explicitPath;
+ }
+
+ /**
+ * Records a resolved path. Returns false only for an identical duplicate entry for the same
+ * spec, which must not duplicate every row in that partition.
+ */
+ boolean validateAndRecord(LinkedHashMap<String, String> spec, Path path) {
+ ResolvedPath resolved = ResolvedPath.of(path);
+ ResolvedPath previousForSpec = pathsBySpec.get(spec);
+ if (previousForSpec != null) {
+ if (previousForSpec.equals(resolved)) {
+ return false;
+ }
+ throw overlappingLocations();
+ }
+
+ if (overlapsOwnedPath(resolved)) {
+ throw overlappingLocations();
+ }
+ pathsBySpec.put(new LinkedHashMap<>(spec), resolved);
+ return true;
+ }
+
+ private boolean overlapsOwnedPath(ResolvedPath path) {
+ OwnershipNode node =
+ ownershipRoots.computeIfAbsent(path.fileSystem(), ignored -> new OwnershipNode());
+ String[] segments = path.pathSegments();
+ for (String segment : segments) {
+ // A terminal node reached before the candidate ends is an existing ancestor.
+ if (node.owned) {
+ return true;
+ }
+ node = node.children.computeIfAbsent(segment, ignored -> new OwnershipNode());
+ }
+ // A terminal final node is equality. Children below it make the candidate an ancestor.
+ if (node.owned || !node.children.isEmpty()) {
+ return true;
+ }
+ node.owned = true;
+ return false;
+ }
+
+ /** Returns the canonical URI representation used on the partition-location wire contract. */
+ public static Path canonicalizeExplicitLocation(String location) {
+ String decoded = location;
+ try {
+ while (true) {
+ validateDecodedLocation(decoded);
+ String next = decodePercentOnce(decoded);
+ if (next.equals(decoded)) {
+ break;
+ }
+ decoded = next;
+ }
+
+ Path path = new Path(decoded);
+ URI uri = path.toUri();
+ String scheme = uri.getScheme();
+ String authority = uri.getAuthority();
+ String uriPath = uri.getPath();
+ if (scheme == null
+ || scheme.isEmpty()
+ || uri.getUserInfo() != null
Review Comment:
**[P1] Do not reject URI forms used by supported filesystems.**
These checks reject `hdfs:///path` because it has no authority, even though Paimon documents that form and HadoopFileIO can resolve it through the catalog Hadoop configuration. They also reject the standard ABFS form `abfs://filesystem@account.dfs.core.windows.net/path`, because `java.net.URI` parses `filesystem` as user info even though it is the Azure filesystem/container name rather than a credential. As a result, explicit partition locations are unusable for common HDFS deployments and standard ADLS Gen2 addresses. Please make validation scheme/FileIO-aware: qualify authorityless HDFS through `CatalogContext`, recognize the structural ABFS authority, and continue rejecting actual embedded credentials.
##########
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeFormatTablePartitionsCommand.scala:
##########
@@ -62,11 +62,17 @@ case class PaimonAnalyzeFormatTablePartitionsCommand(
override def run(sparkSession: SparkSession): Seq[Row] = {
val prefix = leadingPrefix(sparkSession)
- val partitions = v2Table.partitionManager
+ val registeredPartitions = v2Table.partitionManager
Review Comment:
**[P2] Validate the full registry before scoped ANALYZE.**
This query is already prefix-pruned, so checking `location` only on the returned partitions does not validate global path ownership or the authoritative explicit-location count. If selected default partition A has a directory that is also owned by an unselected explicit partition B, this guard passes and `FormatTablePartitionStatsCollector` reads that directory as A, probing explicitly owned data and replacing A statistics with B data. That state is reachable through the current ADD/MSCK registration gap or an inconsistent provider. Please run the same count-aware full-registry preflight used by scans and commits, then apply the ANALYZE prefix and reject selected explicit partitions.
##########
paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java:
##########
@@ -832,6 +952,41 @@ public List<Partition> listPartitions(Identifier identifier) throws TableNotExis
}
}
+ @Override
+ public OptionalLong getExplicitPartitionLocationCount(Identifier identifier)
+ throws TableNotExistException {
+ try {
+ GetTableResponse response = api.getTable(identifier);
+ validateExplicitPartitionLocationCount(response);
+ Long count = response.getExplicitPartitionLocationCount();
+ return count == null || count < 0 ? OptionalLong.empty() : OptionalLong.of(count);
+ } catch (NoSuchResourceException e) {
+ throw new TableNotExistException(identifier);
+ } catch (ForbiddenException e) {
+ throw new TableNoPermissionException(identifier, e);
+ }
+ }
+
+ private void validateExplicitPartitionLocationCount(GetTableResponse response) {
+ Long count = response.getExplicitPartitionLocationCount();
+ if (RESTApi.containsCapability(
+ api.options().get(RESTCatalogInternalOptions.SERVER_CAPABILITIES),
+ RESTApi.FORMAT_TABLE_PARTITION_LOCATION_CAPABILITY)
+ && (count == null || count < 0)) {
Review Comment:
**[P1] Preserve the documented unknown-count fallback.**
The OpenAPI description explicitly allows `explicitPartitionLocationCount` to be omitted, and `Catalog#getExplicitPartitionLocationCount` defines an empty result as a signal to load and validate the full registry. Requiring the count whenever the server advertises the location capability contradicts both contracts: a location-capable provider that cannot supply this optional optimization will make every `getTable` and table-detail conversion fail before any fallback can run. Please accept `null` as `OptionalLong.empty()` and use the existing full-registry path, or advertise count support through a separate capability that makes the field mandatory.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org