>
* as unknown leaves the stored one alone either way, and a report never creates or removes a
* partition row.
*
+ * For an existing partition, omitting {@code path} keeps its location, and naming the
+ * partition's own default directory returns it there without deleting data, which needs
+ * replacement statistics for that partition. Additive statistics are rejected for a Format
+ * Table partition that already has a custom location.
+ *
* @param identifier database name and table name
* @param partitions partitions to be created
* @param ignoreIfExists if false, fail when any partition already exists and apply none of the
@@ -1027,7 +1032,8 @@ public void markDonePartitions(Identifier identifier, List>
* 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
+ * @param partitionOptions options aligned with {@code partitions} by position, or null; a
+ * {@code path} naming the partition's default directory returns it there
* @return the partitions the server created and the ones it already held
*/
public CreatePartitionsResponse createPartitions(
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
index 4e0ec7ab821c..e5260a8dcb5b 100644
--- 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
@@ -40,7 +40,9 @@
*
* Statistics ride along optionally, matched to {@code partitionSpecs} by {@link
* PartitionStatistics#spec()} rather than by position, so they may cover only some of them. Both
- * statistics fields are absent unless the client reports.
+ * statistics fields are absent unless the client reports. Partition options align with {@code
+ * partitionSpecs} by position; naming the partition's own default directory in {@code path} asks
+ * the catalog to put it back there, which needs replacement statistics for that partition.
*/
@JsonIgnoreProperties(ignoreUnknown = true)
public class CreatePartitionsRequest implements RESTRequest {
@@ -148,7 +150,10 @@ public Boolean replaceStatistics() {
return replaceStatistics;
}
- /** Options aligned with partition specs; a null list omits the field. */
+ /**
+ * Options aligned with partition specs; a {@code path} naming the partition's default directory
+ * returns it there. A null list omits the field.
+ */
@JsonGetter(FIELD_PARTITION_OPTIONS)
@Nullable
public List> getPartitionOptions() {
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
index 329abcf67b6c..7e824845dbf9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -1106,6 +1106,11 @@ default void createPartitions(Identifier identifier, List> p
/**
* Create partitions atomically unless existing entries are ignored, with optional statistics
* and position-aligned options.
+ *
+ * For an existing partition, omitting {@code path} keeps its location, and naming the
+ * partition's own default directory returns it there without deleting data, which needs
+ * replacement statistics for that partition. Additive statistics are rejected for a Format
+ * Table partition that already has a custom location. Each call is atomic on its own.
*/
default void createPartitions(
Identifier identifier,
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
index 0fb356966d61..71cdb10123b5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -873,10 +873,12 @@ private List> canonicalizePartitionOptions(
String location = copied.get(PATH.key());
if (location != null) {
try {
+ // A partition location may be the table's own directory, which is how a
+ // request returns a partition there, so what a partition may own is judged
+ // where the table is known rather than here.
copied.put(
PATH.key(),
- FormatTablePartitionPathResolver.canonicalizeCustomLocation(
- location, context)
+ FormatTablePartitionPathResolver.canonicalizeLocation(location, context)
.toString());
} catch (IllegalArgumentException e) {
throw invalidPartitionLocation(identifier, partition, e);
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
index 449ef83ab597..72e23a200ce1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
@@ -161,9 +161,9 @@ public void createPartitions(
@Nullable List statistics,
boolean replaceStatistics,
@Nullable List> partitionOptions) {
- validatePartitionOptions(partitionOptions, partitions);
Map, PartitionStatistics> statisticsBySpec =
validateAndIndexStatistics(statistics, partitions);
+ validatePartitionOptions(partitionOptions, partitions);
if (partitions.isEmpty()) {
return;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
index d7169cfdd538..b823fc462ef3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.format;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
@@ -70,7 +71,6 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
-import java.util.function.Predicate;
import java.util.stream.Collectors;
import static org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition;
@@ -207,10 +207,21 @@ public void commit(List commitMessages) {
}
}
- List validatedPartitions = rejectWritesToCustomLocationPartitions(messages);
+ Set> writtenPartitionSpecs;
+ try {
+ writtenPartitionSpecs = writtenPartitionSpecs(messages);
+ } catch (RuntimeException failure) {
+ markPublishedTargetsToPreserveOnAbort(messages);
+ throw failure;
+ }
+ List targetPartitions =
+ prepareCatalogManagedCommit(messages, writtenPartitionSpecs);
- Set> partitionSpecs = new HashSet<>();
- Set clearedPartitionPaths = new HashSet<>();
+ Set> partitionSpecs = new LinkedHashSet<>();
+ Set> reportTargetSpecs = new LinkedHashSet<>();
+ if (overwrite && partitionManager != null) {
+ targetPartitions.forEach(partition -> reportTargetSpecs.add(partition.spec()));
+ }
Path staticPartitionPath = null;
if (staticPartitions != null && !staticPartitions.isEmpty()) {
@@ -223,15 +234,24 @@ public void commit(List commitMessages) {
staticPartitionPath = partitionPath;
if (staticPartitions.size() == partitionKeys.size()) {
partitionSpecs.add(staticPartitions);
+ reportTargetSpecs.add(staticPartitions);
}
if (overwrite) {
- // A static partition may name only the leading keys, in which case the path
- // is a prefix and the partition directories of the remaining keys sit below.
- clearedPartitionPaths.addAll(
- deletePreviousDataFiles(
- Collections.singletonList(partitionPath),
- partitionKeys.size() - staticPartitions.size(),
- cleanupThreadNum));
+ if (partitionManager != null
+ && staticPartitions.size() < partitionKeys.size()) {
+ // A catalog-managed prefix reaches only its registered descendants. Build
+ // their default paths; a custom path is metadata, never a delete target.
+ deletePreviousDataFiles(
+ tableDataDirectories(targetPartitions, writtenPartitionSpecs),
+ 0,
+ cleanupThreadNum);
+ } else {
+ // A filesystem-discovered prefix is the directory tree underneath it.
+ deletePreviousDataFiles(
+ Collections.singletonList(partitionPath),
+ partitionKeys.size() - staticPartitions.size(),
+ cleanupThreadNum);
+ }
}
} else if (overwrite) {
if (replacesOnlyWrittenPartitions()) {
@@ -248,11 +268,10 @@ public void commit(List commitMessages) {
// Overwriting without naming a partition replaces the table, so what has to go
// is everything the table holds rather than the files this commit happens to
// write: a statement whose query returns nothing still empties the table.
- clearedPartitionPaths.addAll(
- deletePreviousDataFiles(
- tableDataDirectories(validatedPartitions),
- 0,
- cleanupThreadNum));
+ deletePreviousDataFiles(
+ tableDataDirectories(targetPartitions, writtenPartitionSpecs),
+ 0,
+ cleanupThreadNum);
}
}
if (overwrite) {
@@ -280,6 +299,7 @@ public void commit(List commitMessages) {
extractPartitionSpecFromPath(
committer.targetPath().getParent(), partitionKeys);
partitionSpecs.add(spec);
+ reportTargetSpecs.add(spec);
if (reportsStatistics) {
statisticsByPartition.merge(
spec,
@@ -298,12 +318,7 @@ public void commit(List commitMessages) {
message.getCommitter().clean(this.fileIO);
}
if (reportsStatistics && overwrite) {
- reportPartitions(
- partitionSpecs,
- statisticsByPartition,
- clearedPartitionPaths,
- commitTime,
- overwrite);
+ reportPartitions(reportTargetSpecs, statisticsByPartition, commitTime, overwrite);
} else if (partitionManager != null && !partitionSpecs.isEmpty()) {
// Register an append before reporting its additive statistics. Registration is
// idempotent, so a failed multi-batch call can roll back every file from this
@@ -335,12 +350,7 @@ public void commit(List commitMessages) {
markPublishedTargetsToPreserveOnAbort(messages);
if (reportsStatistics && !statisticsByPartition.isEmpty()) {
try {
- reportPartitions(
- partitionSpecs,
- statisticsByPartition,
- clearedPartitionPaths,
- commitTime,
- false);
+ reportPartitions(partitionSpecs, statisticsByPartition, commitTime, false);
} catch (RuntimeException statisticsFailure) {
LOG.warn(
"Committed data for format table {}, but failed to report append "
@@ -397,15 +407,23 @@ private void publishMessages(List messages) throws IOExce
}
}
- /** Rejects writes whose files would belong to a catalog partition outside the table root. */
- private List rejectWritesToCustomLocationPartitions(
- List messages) {
+ /** Loads the registry rows this operation needs before any table mutation. */
+ private List prepareCatalogManagedCommit(
+ List messages, Set> writtenPartitionSpecs) {
if (partitionManager == null || partitionKeys == null || partitionKeys.isEmpty()) {
return Collections.emptyList();
}
try {
- return rejectWritesToCustomLocationPartitionsBeforeMutation(messages);
+ List targetPartitions = loadCommitTargetPartitions(writtenPartitionSpecs);
+ if (!overwrite) {
+ for (Partition partition : targetPartitions) {
+ if (FormatTablePartitionPathResolver.customLocation(partition) != null) {
+ throw unsupportedCustomLocation("Writing", partition);
+ }
+ }
+ }
+ return targetPartitions;
} catch (RuntimeException failure) {
// Nothing has been published yet. Abort should clean staging only: the target may be
// a pre-existing file in a directory owned by another partition.
@@ -414,60 +432,41 @@ private List rejectWritesToCustomLocationPartitions(
}
}
- private List rejectWritesToCustomLocationPartitionsBeforeMutation(
- List messages) {
- Predicate affectsPartition;
- boolean hasStaticPrefixWithoutFiles = false;
- if (overwrite && staticPartitions != null && !staticPartitions.isEmpty()) {
- LinkedHashMap staticSpec = orderedPartitionPrefix(staticPartitions);
- if (staticSpec.size() == partitionKeys.size()) {
- affectsPartition = partition -> partition.spec().equals(staticPartitions);
- } else {
- affectsPartition =
- partition -> partitionSpecMatchesPrefix(partition.spec(), staticSpec);
- }
- } else if (overwrite && !replacesOnlyWrittenPartitions()) {
- affectsPartition = ignored -> true;
- } else {
- Set> affectedSpecs = new LinkedHashSet<>();
- for (TwoPhaseCommitMessage message : messages) {
- Path targetPath = message.getCommitter().targetPath();
- if (targetPath == null) {
- // Preserve the established failure order for a malformed committer. The
- // publish or registration path will report its own contract violation.
- continue;
- }
- affectedSpecs.add(
- extractPartitionSpecFromPath(targetPath.getParent(), partitionKeys));
+ private List loadCommitTargetPartitions(
+ Set> writtenPartitionSpecs) {
+ if (overwrite) {
+ if (staticPartitions == null || staticPartitions.isEmpty()) {
+ return replacesOnlyWrittenPartitions()
+ ? Collections.emptyList()
+ : loadPartitionRegistry();
}
- if (!overwrite && staticPartitions != null && !staticPartitions.isEmpty()) {
- LinkedHashMap staticSpec = orderedPartitionPrefix(staticPartitions);
- if (staticSpec.size() == partitionKeys.size()) {
- if (affectedSpecs.isEmpty()) {
- affectedSpecs.add(staticPartitions);
- }
- } else {
- hasStaticPrefixWithoutFiles = affectedSpecs.isEmpty();
- }
- }
- if (affectedSpecs.isEmpty() && !hasStaticPrefixWithoutFiles) {
- return Collections.emptyList();
- }
- affectsPartition =
- affectedSpecs.isEmpty()
- ? ignored -> false
- : partition -> affectedSpecs.contains(partition.spec());
- }
-
- List registry = loadPartitionRegistry();
- List affectedPartitions =
- registry.stream().filter(affectsPartition).collect(Collectors.toList());
- for (Partition partition : affectedPartitions) {
- if (FormatTablePartitionPathResolver.customLocation(partition) != null) {
- throw unsupportedCustomLocation(overwrite ? "Overwriting" : "Writing", partition);
+ LinkedHashMap staticSpec = orderedPartitionPrefix(staticPartitions);
+ return staticSpec.size() == partitionKeys.size()
+ ? Collections.emptyList()
+ : loadPartitionsByPrefix(staticSpec);
+ }
+
+ return writtenPartitionSpecs.isEmpty()
+ ? Collections.emptyList()
+ : loadPartitionsByNames(writtenPartitionSpecs);
+ }
+
+ private Set> writtenPartitionSpecs(List messages) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptySet();
+ }
+ Set> writtenPartitionSpecs = new LinkedHashSet<>();
+ for (TwoPhaseCommitMessage message : messages) {
+ Path targetPath = message.getCommitter().targetPath();
+ if (targetPath == null) {
+ // Preserve the established failure order for a malformed committer. The publish
+ // or registration path will report its own contract violation.
+ continue;
}
+ writtenPartitionSpecs.add(
+ extractPartitionSpecFromPath(targetPath.getParent(), partitionKeys));
}
- return registry;
+ return writtenPartitionSpecs;
}
private LinkedHashMap orderedPartitionPrefix(
@@ -492,9 +491,42 @@ private LinkedHashMap orderedPartitionPrefix(
return orderedSpec;
}
- /** Validates every registered path before using it for a write or truncate decision. */
+ private List loadPartitionsByNames(Set> partitionSpecs) {
+ List partitions =
+ validatePartitionRegistry(
+ partitionManager.listPartitionsByNames(new ArrayList<>(partitionSpecs)));
+ for (Partition partition : partitions) {
+ if (!partitionSpecs.contains(partition.spec())) {
+ throw new IllegalStateException(
+ String.format(
+ "Catalog returned unrequested partition %s for Format Table %s.",
+ partition.spec(), tableIdentifier.getFullName()));
+ }
+ }
+ return partitions;
+ }
+
+ private List loadPartitionsByPrefix(Map prefix) {
+ List partitions =
+ validatePartitionRegistry(partitionManager.listPartitions(prefix, null));
+ for (Partition partition : partitions) {
+ if (!partitionSpecMatchesPrefix(partition.spec(), prefix)) {
+ throw new IllegalStateException(
+ String.format(
+ "Catalog returned partition %s outside requested prefix %s for Format Table %s.",
+ partition.spec(), prefix, tableIdentifier.getFullName()));
+ }
+ }
+ return partitions;
+ }
+
+ /** Loads and validates the full registry for an operation whose target is the whole table. */
private List loadPartitionRegistry() {
- List partitions = partitionManager.listPartitions(Collections.emptyMap(), null);
+ return validatePartitionRegistry(
+ partitionManager.listPartitions(Collections.emptyMap(), null));
+ }
+
+ private List validatePartitionRegistry(List partitions) {
FormatTablePartitionRegistryValidator.validatePartitionLocations(
partitions,
partitionKeys,
@@ -544,39 +576,46 @@ private static void markPublishedTargetsToPreserveOnAbort(
}
/**
- * Registers the partitions this commit touched, carrying the statistics of what it wrote. An
- * overwrite also empties partitions it writes nothing to - those below a static prefix, and
- * every partition the table has when the statement names none and dynamic partition overwrite
- * is off; those report an exact zero and are registered with the rest, since a statistic can
- * only be reported for a partition its own request registers. A truncation writes nothing and
- * reports every partition it emptied.
+ * Registers every target with the statistics it holds after this operation. A replacement
+ * reports zero for a target it did not write and removes its custom-location metadata in the
+ * same request; the external location itself is never a delete target.
*/
private void reportPartitions(
- Set> writtenPartitionSpecs,
+ Set> targetPartitionSpecs,
Map, PartitionStatistics> statisticsByPartition,
- Set clearedPartitionPaths,
long commitTime,
boolean replaceStatistics) {
- for (Path cleared : clearedPartitionPaths) {
- Map spec = clearedPartitionSpec(cleared);
- if (spec != null) {
- // Emptied and not written to: an exact zero, dated to the commit that did it.
- statisticsByPartition.putIfAbsent(spec, emptyStatistics(spec, commitTime));
- }
- }
-
// Statistics are matched by spec, not by position: the specs need only be a superset.
- Set> specs = new LinkedHashSet<>(writtenPartitionSpecs);
+ Set> specs = new LinkedHashSet<>(targetPartitionSpecs);
specs.addAll(statisticsByPartition.keySet());
if (specs.isEmpty()) {
return;
}
+
+ List> partitionOptions = null;
+ if (replaceStatistics) {
+ // Replacing a partition means owning where it lives: name the default directory, which
+ // returns a partition registered elsewhere and says nothing new about the rest.
+ partitionOptions = new ArrayList<>(specs.size());
+ for (Map spec : specs) {
+ statisticsByPartition.putIfAbsent(spec, emptyStatistics(spec, commitTime));
+ partitionOptions.add(
+ Collections.singletonMap(
+ CoreOptions.PATH.key(),
+ buildPartitionPath(
+ location,
+ spec,
+ formatTablePartitionOnlyValueInPath,
+ partitionKeys)
+ .toString()));
+ }
+ }
partitionManager.createPartitions(
new ArrayList<>(specs),
true,
new ArrayList<>(statisticsByPartition.values()),
replaceStatistics,
- null);
+ partitionOptions);
}
/** What one commit wrote into a partition, with one more of its files folded in. */
@@ -597,60 +636,6 @@ private static long add(long sum, long value) {
: PartitionStatistics.UNKNOWN;
}
- /**
- * The partition a cleared directory belongs to, or null when it is none of this table's.
- * Requiring the spec to rebuild the same directory rules out one nested below a partition,
- * whose trailing components would otherwise read as some other partition; such a directory is
- * left alone, since stale statistics beat statistics of the wrong partition.
- */
- @Nullable
- private Map clearedPartitionSpec(Path clearedPath) {
- LinkedHashMap spec =
- formatTablePartitionOnlyValueInPath
- ? PartitionPathUtils.extractPartitionSpecFromPathOnlyValue(
- clearedPath, partitionKeys)
- : PartitionPathUtils.extractPartitionSpecFromPath(
- clearedPath, partitionKeys);
- if (spec == null) {
- LOG.warn(
- "Cleared directory {} of table {} is not one of its partition directories; "
- + "its partition statistics are left unchanged.",
- clearedPath,
- tableIdentifier.getFullName());
- return null;
- }
- Path rebuilt =
- buildPartitionPath(
- location, spec, formatTablePartitionOnlyValueInPath, partitionKeys);
- if (!samePathComponent(rebuilt, clearedPath)) {
- LOG.warn(
- "Cleared directory {} of table {} does not rebuild from partition spec {}; "
- + "its partition statistics are left unchanged.",
- clearedPath,
- tableIdentifier.getFullName(),
- spec);
- return null;
- }
- return spec;
- }
-
- /**
- * Whether two paths name the same directory, ignoring scheme and authority: a {@link FileIO}
- * that delegates answers a listing under the scheme it used, not the one it was asked with.
- */
- private static boolean samePathComponent(Path left, Path right) {
- return trimTrailingSeparators(left.toUri().normalize().getPath())
- .equals(trimTrailingSeparators(right.toUri().normalize().getPath()));
- }
-
- private static String trimTrailingSeparators(String path) {
- String trimmed = path;
- while (trimmed.length() > 1 && trimmed.endsWith(Path.SEPARATOR)) {
- trimmed = trimmed.substring(0, trimmed.length() - 1);
- }
- return trimmed;
- }
-
private Method getHiveCreatePartitionsInHmsMethod() throws NoSuchMethodException {
Method hiveCreatePartitionsInHmsMethod =
hiveCatalog
@@ -810,21 +795,27 @@ private boolean replacesOnlyWrittenPartitions() {
/**
* The directories this table's data sits in: the table directory itself when the table is
* unpartitioned, and one per partition otherwise, taken from wherever the table reads its
- * partitions. A directory no scan of the table reads holds none of its data - one the catalog
- * has not registered, or one whose name does not parse into the partition keys - and replacing
- * what the table holds leaves it alone, the way {@link #truncateTable()} does.
+ * partitions. A catalog-managed overwrite also includes a written spec not registered yet: it
+ * must clear that default directory before registering it, or unrelated files there would
+ * become visible with the new output.
*/
- private List tableDataDirectories(List validatedPartitions) {
+ private List tableDataDirectories(
+ List targetPartitions, Set> writtenPartitionSpecs) {
if (partitionKeys == null || partitionKeys.isEmpty()) {
return Collections.singletonList(new Path(location));
}
List directories = new ArrayList<>();
if (partitionManager != null) {
- for (Partition partition : validatedPartitions) {
+ Set> targetSpecs = new LinkedHashSet<>();
+ for (Partition partition : targetPartitions) {
+ targetSpecs.add(partition.spec());
+ }
+ targetSpecs.addAll(writtenPartitionSpecs);
+ for (Map targetSpec : targetSpecs) {
directories.add(
buildPartitionPath(
location,
- partition.spec(),
+ targetSpec,
formatTablePartitionOnlyValueInPath,
partitionKeys));
}
@@ -1211,11 +1202,6 @@ public void truncateTable() {
// by whatever the table reads its partitions from.
if (partitionManager != null) {
List partitions = loadPartitionRegistry();
- for (Partition partition : partitions) {
- if (FormatTablePartitionPathResolver.customLocation(partition) != null) {
- throw unsupportedCustomLocation("Truncating", partition);
- }
- }
truncate(partitions.stream().map(Partition::spec).collect(Collectors.toList()));
return;
}
@@ -1248,41 +1234,50 @@ public void truncatePartitions(List> partitionSpecs) {
truncate(normalizedSpecs);
return;
}
- List registry = loadPartitionRegistry();
- Map, Partition> partitions =
- selectRequestedPartitions(registry, normalizedSpecs);
- for (Partition partition : partitions.values()) {
- if (FormatTablePartitionPathResolver.customLocation(partition) != null) {
- throw unsupportedCustomLocation("Truncating", partition);
+ List partitions = loadRequestedPartitions(normalizedSpecs);
+ truncate(partitions.stream().map(Partition::spec).collect(Collectors.toList()));
+ }
+
+ private List loadRequestedPartitions(List> partitionSpecs) {
+ Set> exactSpecs = new LinkedHashSet<>();
+ Set> prefixes = new LinkedHashSet<>();
+ for (Map partitionSpec : partitionSpecs) {
+ if (partitionSpec.size() == partitionKeys.size()) {
+ exactSpecs.add(partitionSpec);
+ } else {
+ prefixes.add(partitionSpec);
}
}
- truncate(partitions.values().stream().map(Partition::spec).collect(Collectors.toList()));
- }
+ if (prefixes.contains(Collections.emptyMap())) {
+ return loadPartitionRegistry();
+ }
+
+ List found = new ArrayList<>();
+ if (!exactSpecs.isEmpty()) {
+ found.addAll(loadPartitionsByNames(exactSpecs));
+ }
+ for (Map prefix : prefixes) {
+ found.addAll(loadPartitionsByPrefix(prefix));
+ }
- private Map, Partition> selectRequestedPartitions(
- List registry, List> partitionSpecs) {
Map, Partition> selected = new LinkedHashMap<>();
- Set> requestedPrefixes = new HashSet<>(partitionSpecs);
- for (Partition partition : registry) {
- if (requestedPrefixes.contains(Collections.emptyMap())) {
- selected.putIfAbsent(partition.spec(), partition);
- continue;
- }
- LinkedHashMap registeredPrefix = new LinkedHashMap<>();
- for (String partitionKey : partitionKeys) {
- registeredPrefix.put(partitionKey, partition.spec().get(partitionKey));
- if (requestedPrefixes.contains(registeredPrefix)) {
- selected.putIfAbsent(partition.spec(), partition);
- break;
- }
+ for (Partition partition : found) {
+ Partition previous = selected.putIfAbsent(partition.spec(), partition);
+ if (previous != null
+ && !Objects.equals(
+ FormatTablePartitionPathResolver.customLocation(previous),
+ FormatTablePartitionPathResolver.customLocation(partition))) {
+ throw new IllegalStateException(
+ String.format(
+ "Catalog returned conflicting locations for partition %s of Format Table %s.",
+ partition.spec(), tableIdentifier.getFullName()));
}
}
- return selected;
+ return validatePartitionRegistry(new ArrayList<>(selected.values()));
}
private void truncate(List> partitionSpecs) {
long truncateTime = System.currentTimeMillis();
- Set clearedPartitionPaths = new HashSet<>();
// Statistics are keyed by the spec that named the partition, so only a complete one can
// seed them; a prefix reaches here only for a table with nowhere to report to.
Map, PartitionStatistics> emptied = new LinkedHashMap<>();
@@ -1295,9 +1290,7 @@ private void truncate(List> partitionSpecs) {
formatTablePartitionOnlyValueInPath,
partitionKeys);
try {
- clearedPartitionPaths.addAll(
- deletePreviousDataFile(
- partitionPath, partitionKeys.size() - partitionSpec.size()));
+ deletePreviousDataFile(partitionPath, partitionKeys.size() - partitionSpec.size());
} catch (Exception e) {
failure =
new RuntimeException(
@@ -1319,11 +1312,7 @@ private void truncate(List> partitionSpecs) {
// too, so the catalog stops describing files that are gone.
try {
reportPartitions(
- Collections.emptySet(),
- emptied,
- clearedPartitionPaths,
- truncateTime,
- /* replaceStatistics */ true);
+ emptied.keySet(), emptied, truncateTime, /* replaceStatistics */ true);
} catch (RuntimeException e) {
if (failure == null) {
throw e;
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
index 0c74250e410a..aebbcfd1fbde 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
@@ -78,7 +78,12 @@ default void createPartitions(List> partitions, boolean igno
* holds or add to it, and is ignored when {@code statistics} is null. A field reported as
* unknown says nothing about itself and leaves the stored one as it was, so a measurement that
* could not take a number does not erase the last one that could. Reporting never unregisters a
- * partition. {@code partitionOptions}, when present, align with {@code partitions} by position.
+ * partition. {@code partitionOptions}, when present, align with {@code partitions} by position;
+ * a {@code path} naming a registered partition's own default directory returns it there and
+ * needs replacement statistics for that partition, and additive statistics are rejected for a
+ * partition that already has a custom location. Implementations validate all arguments first
+ * and may split the request into catalog-sized batches; each batch is atomic, the whole call is
+ * not.
*
* This is the method an implementation provides, so that none can report nothing by
* accident: a decorator that forwards only the two-argument form would otherwise drop every
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionPathResolver.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionPathResolver.java
index 6c5c6327e230..620c1b2690ea 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionPathResolver.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionPathResolver.java
@@ -80,12 +80,8 @@ static String customLocation(Partition partition) {
}
Path resolve(LinkedHashMap spec, @Nullable String customLocation) {
- Path defaultPath =
- new Path(
- tablePath,
- PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath));
if (customLocation == null) {
- return defaultPath;
+ return defaultPartitionPath(tablePath, spec, onlyValueInPath);
}
try {
@@ -96,6 +92,39 @@ Path resolve(LinkedHashMap spec, @Nullable String customLocation
}
}
+ /** Where a partition lives when it carries no location of its own. */
+ public static Path defaultPartitionPath(
+ Path tablePath, LinkedHashMap spec, boolean onlyValueInPath) {
+ return new Path(
+ tablePath, PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath));
+ }
+
+ /**
+ * Whether a requested location names the partition's own default directory. That is how a
+ * request asks for a partition to go back to it, so both sides are compared after
+ * canonicalization and do not have to spell the same directory the same way. A location that
+ * does not parse is not the default one; the caller rejects it on its own terms.
+ */
+ public static boolean isDefaultPartitionPath(
+ Path tablePath,
+ LinkedHashMap spec,
+ boolean onlyValueInPath,
+ String requestedLocation,
+ @Nullable CatalogContext catalogContext) {
+ Path requested;
+ try {
+ PartitionPathUtils.validatePartitionSpecForPath(spec, onlyValueInPath);
+ requested = canonicalizeLocation(requestedLocation, catalogContext);
+ } catch (IllegalArgumentException e) {
+ return false;
+ }
+ return ResolvedPath.of(requested, catalogContext)
+ .equals(
+ ResolvedPath.of(
+ defaultPartitionPath(tablePath, spec, onlyValueInPath),
+ catalogContext));
+ }
+
/** Resolves a custom location using the catalog's Hadoop filesystem identity. */
public static Path resolveCustomLocation(
Path tablePath,
@@ -155,18 +184,44 @@ private boolean overlapsOwnedPath(ResolvedPath path) {
return false;
}
- /** Canonicalizes a custom location using the catalog's Hadoop configuration when present. */
+ /**
+ * Canonicalizes a custom location and requires it to name a place a partition may own: a scheme
+ * that addresses storage the way that storage is addressed.
+ */
public static Path canonicalizeCustomLocation(
String location, @Nullable CatalogContext catalogContext) {
+ // A location someone typed must not smuggle a traversal through an escape, so it is read
+ // once decoded and canonicalized from what it decodes to.
+ validateDecodedLocation(location);
+ String decoded = decodePercentOnce(location);
+ if (decoded.contains("%")) {
+ throw new IllegalArgumentException("Invalid custom partition location.");
+ }
+ Path canonical = canonicalizeLocation(decoded, catalogContext);
+ URI uri = canonical.toUri();
+ String scheme = uri.getScheme();
+ String authority = uri.getAuthority();
+ if ((scheme.equals("file") && authority != null && !authority.isEmpty())
+ || (!scheme.equals("file")
+ && !scheme.equals("hdfs")
+ && (authority == null || authority.isEmpty()))) {
+ throw new IllegalArgumentException("Invalid custom partition location.");
+ }
+ return canonical;
+ }
+
+ /**
+ * Canonicalizes any location the catalog carries, including a table's own directory: its scheme
+ * may address storage without an authority, and a partition directory keeps the escapes that
+ * make its name one name, so {@code dt=a%2Fb} is not the two levels {@code dt=a} and {@code b}.
+ * Callers that know the table judge whether the result is a place a partition may own.
+ */
+ public static Path canonicalizeLocation(
+ String location, @Nullable CatalogContext catalogContext) {
try {
validateDecodedLocation(location);
- String decoded = decodePercentOnce(location);
- if (decoded.contains("%")) {
- throw new IllegalArgumentException("Invalid custom partition location.");
- }
- validateDecodedLocation(decoded);
- Path path = new Path(decoded);
+ Path path = new Path(location);
URI uri = path.toUri();
String scheme = uri.getScheme();
String authority = uri.getAuthority();
@@ -181,12 +236,6 @@ public static Path canonicalizeCustomLocation(
}
scheme = scheme.toLowerCase(Locale.ROOT);
- if ((scheme.equals("file") && authority != null && !authority.isEmpty())
- || (!scheme.equals("file")
- && !scheme.equals("hdfs")
- && (authority == null || authority.isEmpty()))) {
- throw new IllegalArgumentException("Invalid custom partition location.");
- }
authority =
authority == null || authority.isEmpty()
? null
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
index 210db841ebb9..492b509dfd1f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
@@ -591,7 +591,7 @@ void testRenamePreservesCustomPartitionLocation() throws Exception {
}
@Test
- void testInvalidCustomPartitionLocationFailsBeforePost() throws Exception {
+ void testInvalidCustomPartitionLocationIsRejectedWithoutMutation() throws Exception {
Identifier identifier = createFormatTableWithCatalogManagedPartitions();
Map spec = Collections.singletonMap("dt", "20260717");
String partitionsResource =
@@ -609,10 +609,12 @@ void testInvalidCustomPartitionLocationFailsBeforePost() throws Exception {
false,
partitionOptions(
"oss://archive-bucket/history/%2e%2e/secret")))
- .isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Invalid custom partition location");
- assertThat(restCatalogServer.getReceivedHeaders(partitionsResource)).isEmpty();
+ // The catalog carries the location as written, since only the server knows whether it is
+ // the partition's own default directory; a location that is not is refused there.
+ assertThat(restCatalogServer.getReceivedHeaders(partitionsResource)).hasSize(1);
+ assertThat(restCatalog.listPartitions(identifier)).isEmpty();
}
@Test
@@ -685,7 +687,7 @@ void testPartitionOptionsRejectNullValuesBeforeMutation() throws Exception {
Collections.singletonList(options)),
restCatalog.api().authFunction()))
.isInstanceOf(BadRequestException.class)
- .hasMessageContaining("null keys or values");
+ .hasMessageContaining("partitionOptions must not contain null keys or values");
assertThat(restCatalog.listPartitions(identifier)).isEmpty();
}
@@ -749,6 +751,349 @@ void testExistingPartitionRejectsADifferentCustomLocationWithoutMutation() throw
.containsExactly(tuple(spec, originalLocation));
}
+ @Test
+ void testExistingPartitionAcceptsTheSameCustomLocationIdempotently() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ Map originalOptions = new HashMap<>();
+ originalOptions.put(CoreOptions.PATH.key(), "file:/archive/original");
+ originalOptions.put("owner", "data-platform");
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ Collections.singletonList(originalOptions));
+
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ partitionOptions("file:/archive/original"));
+
+ // An idempotent retry with the same path must preserve unrelated stored options.
+ assertThat(onlyPartition(identifier).options())
+ .containsExactlyInAnyOrderEntriesOf(originalOptions);
+ }
+
+ @Test
+ void testReplacementStatisticsWithoutALocationPreserveCustomLocation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ Map originalOptions = new HashMap<>();
+ originalOptions.put(CoreOptions.PATH.key(), "file:/archive/original");
+ originalOptions.put("owner", "data-platform");
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(partitionStatistics(spec, 9L)),
+ true,
+ Collections.singletonList(originalOptions));
+
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(partitionStatistics(spec, 3L)),
+ true,
+ null);
+
+ Partition partition = onlyPartition(identifier);
+ assertThat(partition.recordCount()).isEqualTo(3L);
+ // ANALYZE has this request shape. Keying reset off replaceStatistics alone would silently
+ // move an analyzed external partition back under the table directory.
+ assertThat(partition.options()).containsExactlyInAnyOrderEntriesOf(originalOptions);
+
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(partitionStatistics(spec, 4L)),
+ true,
+ Collections.singletonList(
+ Collections.singletonMap("owner", "must-not-overwrite-existing-options")));
+
+ partition = onlyPartition(identifier);
+ assertThat(partition.recordCount()).isEqualTo(4L);
+ // An aligned non-path option is not a path reset and does not patch stored options.
+ assertThat(partition.options()).containsExactlyInAnyOrderEntriesOf(originalOptions);
+ }
+
+ @Test
+ void testReturningToTheDefaultDirectoryRemovesOnlyLocationAndIsReplaySafe() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ Map newSpec = Collections.singletonMap("dt", "20260718");
+ String releasedLocation = "file:/archive/original";
+ Map originalOptions = new HashMap<>();
+ originalOptions.put(CoreOptions.PATH.key(), releasedLocation);
+ originalOptions.put("owner", "data-platform");
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(partitionStatistics(spec, 9L)),
+ true,
+ Collections.singletonList(originalOptions));
+
+ List> specs = Arrays.asList(spec, newSpec);
+ List replacement =
+ Arrays.asList(partitionStatistics(spec, 3L), partitionStatistics(newSpec, 7L));
+ Map resetOptions = returnToDefault(identifier, spec);
+ resetOptions.put("owner", "must-not-overwrite-existing-options");
+ List> resetAndReuse =
+ Arrays.asList(
+ resetOptions,
+ Collections.singletonMap(CoreOptions.PATH.key(), releasedLocation));
+ restCatalog.createPartitions(identifier, specs, true, replacement, true, resetAndReuse);
+
+ assertThat(restCatalog.listPartitions(identifier))
+ .extracting(Partition::spec, Partition::recordCount, Partition::options)
+ .containsExactlyInAnyOrder(
+ tuple(spec, 3L, Collections.singletonMap("owner", "data-platform")),
+ tuple(
+ newSpec,
+ 7L,
+ Collections.singletonMap(
+ CoreOptions.PATH.key(), releasedLocation)));
+
+ restCatalog.createPartitions(identifier, specs, true, replacement, true, resetAndReuse);
+
+ // Replaying the final-state transition replaces, rather than adds, both statistics rows.
+ assertThat(restCatalog.listPartitions(identifier))
+ .extracting(Partition::spec, Partition::recordCount, Partition::options)
+ .containsExactlyInAnyOrder(
+ tuple(spec, 3L, Collections.singletonMap("owner", "data-platform")),
+ tuple(
+ newSpec,
+ 7L,
+ Collections.singletonMap(
+ CoreOptions.PATH.key(), releasedLocation)));
+ }
+
+ @Test
+ void testNewPartitionAtItsDefaultDirectoryStoresNoLocation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map ordinarySpec = Collections.singletonMap("dt", "20260717");
+ Map resetSpec = Collections.singletonMap("dt", "20260718");
+ restCatalog.createPartitions(identifier, Collections.singletonList(ordinarySpec));
+
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(resetSpec),
+ true,
+ Collections.singletonList(partitionStatistics(resetSpec, 7L)),
+ true,
+ Collections.singletonList(returnToDefault(identifier, resetSpec)));
+
+ // Naming the default directory is an instruction, never stored data: the new row uses
+ // exactly the same options representation as a normal default-location registration.
+ assertThat(restCatalog.listPartitions(identifier))
+ .extracting(Partition::spec, Partition::recordCount, Partition::options)
+ .containsExactlyInAnyOrder(
+ tuple(ordinarySpec, PartitionStatistics.UNKNOWN, null),
+ tuple(resetSpec, 7L, null));
+ }
+
+ @Test
+ void testAdditiveStatisticsOnCustomPartitionRejectsWholeBatchWithoutMutation()
+ throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map ordinary = Collections.singletonMap("dt", "20260717");
+ Map custom = Collections.singletonMap("dt", "20260718");
+ Map createdByRejectedBatch = Collections.singletonMap("dt", "20260719");
+ restCatalog.createPartitions(
+ identifier,
+ Arrays.asList(ordinary, custom),
+ true,
+ Arrays.asList(partitionStatistics(ordinary, 10L), partitionStatistics(custom, 20L)),
+ true,
+ Arrays.asList(
+ Collections.emptyMap(),
+ Collections.singletonMap(CoreOptions.PATH.key(), "file:/archive/custom")));
+ List before = restCatalog.listPartitions(identifier);
+
+ assertThatThrownBy(
+ () ->
+ restCatalog.createPartitions(
+ identifier,
+ Arrays.asList(ordinary, createdByRejectedBatch, custom),
+ true,
+ Arrays.asList(
+ partitionStatistics(ordinary, 1L),
+ partitionStatistics(createdByRejectedBatch, 1L),
+ partitionStatistics(custom, 1L)),
+ false,
+ null))
+ .isInstanceOf(IllegalArgumentException.class);
+
+ // The ordinary row comes first so an implementation cannot mutate while walking the
+ // request and only then discover the custom row at the end.
+ assertThat(restCatalog.listPartitions(identifier)).containsExactlyElementsOf(before);
+ }
+
+ @Test
+ void testPostValidationFailureRollsBackPathStatisticsAndRegistration() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map resetSpec = Collections.singletonMap("dt", "20260717");
+ Map stableSpec = Collections.singletonMap("dt", "20260718");
+ Map newSpec = Collections.singletonMap("dt", "20260719");
+ Map resetOptions = new HashMap<>();
+ resetOptions.put(CoreOptions.PATH.key(), "file:/archive/reset-target");
+ resetOptions.put("owner", "data-platform");
+ restCatalog.createPartitions(
+ identifier,
+ Arrays.asList(resetSpec, stableSpec),
+ true,
+ Arrays.asList(
+ partitionStatistics(resetSpec, 10L), partitionStatistics(stableSpec, 20L)),
+ true,
+ Arrays.asList(
+ resetOptions,
+ Collections.singletonMap(CoreOptions.PATH.key(), "file:/archive/stable")));
+ List before = restCatalog.listPartitions(identifier);
+ String resource =
+ ResourcePaths.forCatalogProperties(restCatalog.api().options())
+ .partitions(identifier.getDatabaseName(), identifier.getObjectName());
+ restCatalogServer.clearReceivedHeaders();
+
+ assertThatThrownBy(
+ () ->
+ restCatalog.createPartitions(
+ identifier,
+ Arrays.asList(resetSpec, newSpec),
+ true,
+ Arrays.asList(
+ partitionStatistics(resetSpec, 1L),
+ partitionStatistics(newSpec, 1L)),
+ true,
+ Arrays.asList(
+ returnToDefault(identifier, resetSpec),
+ Collections.singletonMap(
+ CoreOptions.PATH.key(),
+ "file:/archive/stable"))))
+ .isInstanceOf(IllegalArgumentException.class);
+
+ // This must be a server-side prospective-registry rejection, not a client-side shortcut.
+ assertThat(restCatalogServer.getReceivedHeaders(resource)).hasSize(1);
+ // Partition equality covers all statistics, audit fields, done state, and every option.
+ assertThat(restCatalog.listPartitions(identifier)).containsExactlyElementsOf(before);
+ }
+
+ @Test
+ void testAReturnToDefaultNeedsItsOwnStatisticsNotAnotherPartitionsAtomically()
+ throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map resetSpec = Collections.singletonMap("dt", "20260717");
+ Map otherSpec = Collections.singletonMap("dt", "20260718");
+ Map originalOptions = new HashMap<>();
+ originalOptions.put(CoreOptions.PATH.key(), "file:/archive/reset-target");
+ originalOptions.put("owner", "data-platform");
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(resetSpec),
+ true,
+ Collections.singletonList(partitionStatistics(resetSpec, 10L)),
+ true,
+ Collections.singletonList(originalOptions));
+ List before = restCatalog.listPartitions(identifier);
+ String resource =
+ ResourcePaths.forCatalogProperties(restCatalog.api().options())
+ .partitions(identifier.getDatabaseName(), identifier.getObjectName());
+ HttpClient client = new HttpClient(restCatalogServer.getUrl());
+ restCatalogServer.clearReceivedHeaders();
+
+ assertThatThrownBy(
+ () ->
+ client.post(
+ resource,
+ new RawCreatePartitionsRequest(
+ Arrays.asList(resetSpec, otherSpec),
+ Collections.singletonList(
+ partitionStatistics(otherSpec, 99L)),
+ true,
+ Arrays.asList(
+ returnToDefault(identifier, resetSpec),
+ Collections.emptyMap())),
+ restCatalog.api().authFunction()))
+ .isInstanceOf(BadRequestException.class)
+ .hasMessageContaining("20260717");
+
+ // Sending the raw request bypasses client validation. The server must reject it without
+ // changing the path, statistics, or audit fields.
+ assertThat(restCatalogServer.getReceivedHeaders(resource)).hasSize(1);
+ assertThat(restCatalog.listPartitions(identifier)).containsExactlyElementsOf(before);
+ }
+
+ @Test
+ void testRawReturnToDefaultWithoutReplacementReportIsRejectedBeforeMutation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map resetSpec = Collections.singletonMap("dt", "20260717");
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(resetSpec),
+ true,
+ Collections.singletonList(partitionStatistics(resetSpec, 10L)),
+ true,
+ partitionOptions("file:/archive/reset-target"));
+ List before = restCatalog.listPartitions(identifier);
+ String resource =
+ ResourcePaths.forCatalogProperties(restCatalog.api().options())
+ .partitions(identifier.getDatabaseName(), identifier.getObjectName());
+ HttpClient client = new HttpClient(restCatalogServer.getUrl());
+ restCatalogServer.clearReceivedHeaders();
+
+ assertThatThrownBy(
+ () ->
+ client.post(
+ resource,
+ new RawCreatePartitionsRequest(
+ Collections.singletonList(resetSpec),
+ null,
+ null,
+ Collections.singletonList(
+ returnToDefault(identifier, resetSpec))),
+ restCatalog.api().authFunction()))
+ .isInstanceOf(BadRequestException.class);
+
+ assertThat(restCatalogServer.getReceivedHeaders(resource)).hasSize(1);
+ assertThat(restCatalog.listPartitions(identifier)).containsExactlyElementsOf(before);
+ }
+
+ @Test
+ void testUnsupportedProviderRejectsReturnToDefaultWithoutMutation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(partitionStatistics(spec, 10L)),
+ true,
+ partitionOptions("file:/archive/reset-target"));
+ List before = restCatalog.listPartitions(identifier);
+ restCatalogServer.setPartitionOptionsCreateSupported(false);
+
+ assertThatThrownBy(
+ () ->
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(partitionStatistics(spec, 0L)),
+ true,
+ Collections.singletonList(
+ returnToDefault(identifier, spec))))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("does not support partition options");
+
+ assertThat(restCatalog.listPartitions(identifier)).containsExactlyElementsOf(before);
+ }
+
@Test
void testConcurrentCustomLocationCreatesValidateAndCommitSerially() throws Exception {
Identifier identifier = createFormatTableWithCatalogManagedPartitions();
@@ -897,10 +1242,7 @@ void testReportedPartitionStatisticsAreStoredAndReadBack() throws Exception {
Identifier identifier = createFormatTableWithCatalogManagedPartitions();
Map spec = Collections.singletonMap("dt", "20260717");
List> specs = Collections.singletonList(spec);
- String location = "file:/archive/dt=20260717";
- Map options = new HashMap<>();
- options.put(CoreOptions.PATH.key(), location);
- options.put("owner", "data-platform");
+ Map options = Collections.singletonMap("owner", "data-platform");
FormatTablePartitionManager partitionManager =
((FormatTable) restCatalog.getTable(identifier)).partitionManager();
assertThat(partitionManager).isNotNull();
@@ -1058,6 +1400,33 @@ private static List> partitionOptions(String... locations) {
return options;
}
+ /** The request that asks for a partition to live at its own default directory again. */
+ private Map returnToDefault(Identifier identifier, Map spec)
+ throws Exception {
+ Map options = new HashMap<>();
+ options.put(CoreOptions.PATH.key(), defaultPartitionDirectory(identifier, spec));
+ return options;
+ }
+
+ private String defaultPartitionDirectory(Identifier identifier, Map spec)
+ throws Exception {
+ FormatTable table = (FormatTable) restCatalog.getTable(identifier);
+ return new org.apache.paimon.fs.Path(
+ new org.apache.paimon.fs.Path(table.location()), "dt=" + spec.get("dt"))
+ .toString();
+ }
+
+ private static PartitionStatistics partitionStatistics(
+ Map spec, long recordCount) {
+ return new PartitionStatistics(
+ spec,
+ recordCount,
+ recordCount * 100,
+ recordCount == 0 ? 0 : 1,
+ 1756684800000L,
+ PartitionStatistics.UNKNOWN_TOTAL_BUCKETS);
+ }
+
private static String customLocation(Partition partition) {
return partition.options() == null ? null : partition.options().get(CoreOptions.PATH.key());
}
@@ -1473,12 +1842,24 @@ private RESTCatalog initCatalogUtil(
private static class RawCreatePartitionsRequest implements RESTRequest {
private final List> partitionSpecs;
+ private final List partitionStatistics;
+ private final Boolean replaceStatistics;
private final List> partitionOptions;
private RawCreatePartitionsRequest(
List> partitionSpecs,
List> partitionOptions) {
+ this(partitionSpecs, null, null, partitionOptions);
+ }
+
+ private RawCreatePartitionsRequest(
+ List> partitionSpecs,
+ List partitionStatistics,
+ Boolean replaceStatistics,
+ List> partitionOptions) {
this.partitionSpecs = partitionSpecs;
+ this.partitionStatistics = partitionStatistics;
+ this.replaceStatistics = replaceStatistics;
this.partitionOptions = partitionOptions;
}
@@ -1492,6 +1873,16 @@ public List> getPartitionOptions() {
return partitionOptions;
}
+ @JsonGetter("partitionStatistics")
+ public List getPartitionStatistics() {
+ return partitionStatistics;
+ }
+
+ @JsonGetter("replaceStatistics")
+ public Boolean replaceStatistics() {
+ return replaceStatistics;
+ }
+
@JsonGetter("ignoreIfExists")
public boolean ignoreIfExists() {
return true;
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTApiJsonTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTApiJsonTest.java
index 9641cf02f0d8..e16233ece532 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTApiJsonTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTApiJsonTest.java
@@ -360,7 +360,41 @@ public void createPartitionsRequestPreservesOptionsTest() throws Exception {
}
@Test
- public void createPartitionsRequestRejectsNullOptionMapsTest() {
+ public void createPartitionsRequestCarriesPartitionLocationsTest() throws Exception {
+ Map returningSpec = Collections.singletonMap("dt", "20260901");
+ Map untouchedSpec = Collections.singletonMap("dt", "20260902");
+ Map defaultDirectory =
+ Collections.singletonMap("path", "file:/warehouse/db/table/dt=20260901");
+ PartitionStatistics replacement =
+ new PartitionStatistics(returningSpec, 0L, 0L, 0L, 1756684800000L, -1);
+ CreatePartitionsRequest request =
+ new CreatePartitionsRequest(
+ Arrays.asList(returningSpec, untouchedSpec),
+ true,
+ Collections.singletonList(replacement),
+ true,
+ Arrays.asList(defaultDirectory, Collections.emptyMap()));
+
+ String json = RESTApi.toJson(request);
+ CreatePartitionsRequest parsed = RESTApi.fromJson(json, CreatePartitionsRequest.class);
+ Map wireObject = RESTApi.fromJson(json, Map.class);
+
+ // Naming the partition's own default directory is how a request asks for it back; an
+ // absent path leaves the stored location alone.
+ assertTrue(json.contains("\"path\":\"file:/warehouse/db/table/dt=20260901\""));
+ assertEquals(
+ Arrays.asList(defaultDirectory, Collections.emptyMap()),
+ parsed.getPartitionOptions());
+ assertEquals(5, wireObject.size());
+ assertTrue(wireObject.containsKey("partitionSpecs"));
+ assertTrue(wireObject.containsKey("ignoreIfExists"));
+ assertTrue(wireObject.containsKey("partitionStatistics"));
+ assertTrue(wireObject.containsKey("replaceStatistics"));
+ assertTrue(wireObject.containsKey("partitionOptions"));
+ }
+
+ @Test
+ public void createPartitionsRequestRejectsNullMapsKeysAndValuesTest() {
List> specs =
Arrays.asList(
Collections.singletonMap("dt", "20260901"),
@@ -377,6 +411,7 @@ public void createPartitionsRequestRejectsNullOptionMapsTest() {
Arrays.asList(null, Collections.emptyMap())));
Map nullValue = new HashMap<>();
+ nullValue.put("path", null);
nullValue.put("owner", null);
assertThrows(
IllegalArgumentException.class,
@@ -384,8 +419,10 @@ public void createPartitionsRequestRejectsNullOptionMapsTest() {
new CreatePartitionsRequest(
specs,
true,
- null,
- null,
+ Collections.singletonList(
+ new PartitionStatistics(
+ specs.get(1), 0L, 0L, 0L, 1756684800000L, -1)),
+ true,
Arrays.asList(Collections.emptyMap(), nullValue)));
Map nullKey = new HashMap<>();
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java
index a7036a9ee97f..28de8abd30ff 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java
@@ -19,6 +19,7 @@
package org.apache.paimon.rest;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.TableType;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.TableMetadata;
import org.apache.paimon.fs.Path;
@@ -36,6 +37,7 @@
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -94,9 +96,12 @@ static List> canonicalizeRequestedOptions(
Map copied = new HashMap<>(partitionOptions);
String location = copied.get(PATH.key());
if (location != null) {
+ // What a partition may own is judged once the location is known not to be the
+ // partition's own default directory, which validateFormatTablePartitionLocations
+ // does after the returns have been taken out.
copied.put(
PATH.key(),
- FormatTablePartitionPathResolver.canonicalizeCustomLocation(
+ FormatTablePartitionPathResolver.canonicalizeLocation(
location, catalogContext)
.toString());
}
@@ -141,7 +146,7 @@ static ErrorResponse conflictingLocationError(Map spec) {
static Partition newPartition(Map spec, @Nullable Map options) {
return new Partition(
- spec,
+ new LinkedHashMap<>(spec),
PartitionStatistics.UNKNOWN,
PartitionStatistics.UNKNOWN,
PartitionStatistics.UNKNOWN,
@@ -152,10 +157,228 @@ static Partition newPartition(Map spec, @Nullable Map copyPartitions(List partitions) {
+ List copied = new ArrayList<>(partitions.size());
+ for (Partition partition : partitions) {
+ copied.add(copyPartition(partition, copyOptions(partition.options())));
+ }
+ return copied;
+ }
+
+ private static void validateNoAdditiveStatisticsForCustomPartitions(
+ List stored,
+ @Nullable List statistics,
+ @Nullable Boolean replaceStatistics) {
+ if (statistics == null || statistics.isEmpty() || Boolean.TRUE.equals(replaceStatistics)) {
+ return;
+ }
+ Set> reportedSpecs = new HashSet<>();
+ for (PartitionStatistics statistic : statistics) {
+ reportedSpecs.add(statistic.spec());
+ }
+ for (Partition partition : stored) {
+ if (customLocation(partition) != null && reportedSpecs.contains(partition.spec())) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot add statistics to custom-location partition %s; "
+ + "reset its path with replacement statistics first.",
+ PartitionUtils.buildPartitionName(partition.spec())));
+ }
+ }
+ }
+
+ /**
+ * Specs whose requested location names their own default directory. Such a request asks for the
+ * partition to live there again, so the location itself is dropped from the request: a
+ * partition at its default directory carries no location of its own.
+ */
+ static Set> takeReturnsToDefault(
+ CreatePartitionsRequest request,
+ @Nullable List> requestedOptions,
+ List stored,
+ TableMetadata metadata,
+ String tableName,
+ CatalogContext catalogContext) {
+ if (!TableType.FORMAT_TABLE
+ .toString()
+ .equalsIgnoreCase(metadata.schema().options().get(CoreOptions.TYPE.key()))) {
+ canonicalizeRemainingLocations(requestedOptions, catalogContext);
+ return java.util.Collections.emptySet();
+ }
+ validateNoAdditiveStatisticsForCustomPartitions(
+ stored, request.getPartitionStatistics(), request.replaceStatistics());
+ Set> returning =
+ defaultDirectoryRequests(
+ request.getPartitionSpecs(),
+ requestedOptions,
+ metadata,
+ tableName,
+ catalogContext);
+ validateReturnsToDefault(
+ returning, request.getPartitionStatistics(), request.replaceStatistics());
+ canonicalizeRemainingLocations(requestedOptions, catalogContext);
+ return returning;
+ }
+
+ /**
+ * A location still named after the returns were taken out is one a partition wants to own, so
+ * it is held to the rules for such a place and stored the way they canonicalize it.
+ */
+ private static void canonicalizeRemainingLocations(
+ @Nullable List> requestedOptions, CatalogContext catalogContext) {
+ if (requestedOptions == null) {
+ return;
+ }
+ for (Map options : requestedOptions) {
+ String location = options.get(PATH.key());
+ if (location != null) {
+ options.put(
+ PATH.key(),
+ FormatTablePartitionPathResolver.canonicalizeCustomLocation(
+ location, catalogContext)
+ .toString());
+ }
+ }
+ }
+
+ private static Set> defaultDirectoryRequests(
+ List> requestedSpecs,
+ @Nullable List> requestedOptions,
+ TableMetadata metadata,
+ String tableName,
+ CatalogContext catalogContext) {
+ if (requestedOptions == null
+ || requestedOptions.stream()
+ .noneMatch(options -> options.get(PATH.key()) != null)) {
+ return java.util.Collections.emptySet();
+ }
+ String tablePath = metadata.schema().options().get(PATH.key());
+ if (StringUtils.isBlank(tablePath)) {
+ throw new IllegalStateException(
+ String.format("Format Table %s has no authoritative path.", tableName));
+ }
+ List partitionKeys = metadata.schema().partitionKeys();
+ boolean onlyValueInPath =
+ new CoreOptions(metadata.schema().options()).formatTablePartitionOnlyValueInPath();
+ Set> returning = new HashSet<>();
+ for (int i = 0; i < requestedOptions.size(); i++) {
+ Map options = requestedOptions.get(i);
+ String requested = options.get(PATH.key());
+ Map spec = requestedSpecs.get(i);
+ if (requested == null || spec == null || !spec.keySet().containsAll(partitionKeys)) {
+ continue;
+ }
+ LinkedHashMap orderedSpec = new LinkedHashMap<>();
+ for (String partitionKey : partitionKeys) {
+ orderedSpec.put(partitionKey, spec.get(partitionKey));
+ }
+ if (FormatTablePartitionPathResolver.isDefaultPartitionPath(
+ new Path(tablePath), orderedSpec, onlyValueInPath, requested, catalogContext)) {
+ options.remove(PATH.key());
+ returning.add(spec);
+ }
+ }
+ return returning;
+ }
+
+ /** Returning a partition to its default directory replaces whatever it held there. */
+ private static void validateReturnsToDefault(
+ Set> returning,
+ @Nullable List statistics,
+ @Nullable Boolean replaceStatistics) {
+ if (returning.isEmpty()) {
+ return;
+ }
+ Set> reportedSpecs = new HashSet<>();
+ if (statistics != null) {
+ for (PartitionStatistics statistic : statistics) {
+ reportedSpecs.add(statistic.spec());
+ }
+ }
+ for (Map spec : returning) {
+ if (!Boolean.TRUE.equals(replaceStatistics) || !reportedSpecs.contains(spec)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Returning partition %s to its default directory requires "
+ + "replaceStatistics=true and statistics for the same partition.",
+ PartitionUtils.buildPartitionName(spec)));
+ }
+ }
+ }
+
+ /** Drops the location of every partition back at its default directory, then checks them. */
+ static void settlePartitionLocations(
+ List partitions,
+ Set> returningToDefault,
+ boolean formatTable,
+ TableMetadata metadata,
+ String tableName,
+ CatalogContext catalogContext) {
+ applyReturnsToDefault(partitions, returningToDefault);
+ if (formatTable) {
+ validateFormatTablePartitionLocations(partitions, metadata, tableName, catalogContext);
+ }
+ }
+
+ private static void applyReturnsToDefault(
+ List partitions, Set> returning) {
+ if (returning.isEmpty()) {
+ return;
+ }
+ for (int i = 0; i < partitions.size(); i++) {
+ Partition partition = partitions.get(i);
+ if (returning.contains(partition.spec())) {
+ partitions.set(i, copyPartition(partition, withoutPath(partition.options())));
+ }
+ }
+ }
+
+ @Nullable
+ private static Map normalizeNewPartitionOptions(
+ @Nullable Map options) {
+ Map copied = copyOptions(options);
+ if (copied == null) {
+ return null;
+ }
+ return copied.isEmpty() ? null : copied;
+ }
+
+ @Nullable
+ private static Map withoutPath(@Nullable Map options) {
+ Map copied = copyOptions(options);
+ if (copied == null) {
+ return null;
+ }
+ copied.remove(PATH.key());
+ return copied.isEmpty() ? null : copied;
+ }
+
+ @Nullable
+ private static Map copyOptions(@Nullable Map options) {
+ return options == null ? null : new HashMap<>(options);
+ }
+
+ private static Partition copyPartition(
+ Partition partition, @Nullable Map options) {
+ return new Partition(
+ new LinkedHashMap<>(partition.spec()),
+ partition.recordCount(),
+ partition.fileSizeInBytes(),
+ partition.fileCount(),
+ partition.lastFileCreationTime(),
+ partition.totalBuckets(),
+ partition.done(),
+ partition.createdAt(),
+ partition.createdBy(),
+ partition.updatedAt(),
+ partition.updatedBy(),
options);
}
- static void validateFormatTablePartitionLocations(
+ private static void validateFormatTablePartitionLocations(
List partitions,
TableMetadata metadata,
String tableName,
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
index 90a948395b11..d196e3be2389 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -2213,7 +2213,7 @@ private MockResponse partitionsApiHandle(
throw new Catalog.TableNotExistException(tableIdentifier);
}
List storedPartitions =
- new ArrayList<>(
+ RESTCatalogPartitionSupport.copyPartitions(
tablePartitionsStore.getOrDefault(
tableName, Collections.emptyList()));
Set> existingSpecs =
@@ -2237,6 +2237,15 @@ private MockResponse partitionsApiHandle(
return mockResponse(response, 409);
}
}
+ boolean formatTable = isFormatTable(tableMetadata.schema().toSchema());
+ Set> returningToDefault =
+ RESTCatalogPartitionSupport.takeReturnsToDefault(
+ request,
+ requestedOptions,
+ storedPartitions,
+ tableMetadata,
+ tableName,
+ catalogContext);
Optional> conflictingLocation =
RESTCatalogPartitionSupport.conflictingLocation(
storedPartitions, request.getPartitionSpecs(), requestedOptions);
@@ -2261,10 +2270,13 @@ private MockResponse partitionsApiHandle(
existed.add(spec);
}
}
- if (isFormatTable(tableMetadata.schema().toSchema())) {
- RESTCatalogPartitionSupport.validateFormatTablePartitionLocations(
- storedPartitions, tableMetadata, tableName, catalogContext);
- }
+ RESTCatalogPartitionSupport.settlePartitionLocations(
+ storedPartitions,
+ returningToDefault,
+ formatTable,
+ tableMetadata,
+ tableName,
+ catalogContext);
applyPartitionStatistics(
storedPartitions,
request.getPartitionStatistics(),
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java
index 3c51dbdc6430..d313c01d5cd7 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java
@@ -38,6 +38,7 @@
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -428,6 +429,91 @@ void testPartitionOptionsStayAlignedAcrossBatches() throws Exception {
assertThat(flatten(optionCaptor.getAllValues())).isEqualTo(options);
}
+ @Test
+ void testPartitionLocationsStayWithReplacementStatisticsAcrossBatches() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ List> specs = specs(2001);
+ List> options = new ArrayList<>(specs.size());
+ List statistics = new ArrayList<>(specs.size());
+ for (int i = 0; i < specs.size(); i++) {
+ if (i == 999 || i == 1000 || i == 2000) {
+ options.add(Collections.singletonMap("path", "file:/warehouse/archive/" + i));
+ } else {
+ options.add(Collections.emptyMap());
+ }
+ }
+ // The partitions carrying a location straddle both split points. Reports are deliberately
+ // reversed: options align by original position while statistics align by spec, so sharing
+ // either indexing rule between them would move a location onto the wrong partition.
+ for (int i = specs.size() - 1; i >= 0; i--) {
+ statistics.add(statistics(specs.get(i), i));
+ }
+
+ partitionManager(catalog).createPartitions(specs, true, statistics, true, options);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor>> specCaptor = ArgumentCaptor.forClass(List.class);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor> statisticsCaptor =
+ ArgumentCaptor.forClass(List.class);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor>> optionCaptor =
+ ArgumentCaptor.forClass(List.class);
+ verify(catalog, times(3))
+ .createPartitions(
+ eq(IDENTIFIER),
+ specCaptor.capture(),
+ eq(true),
+ statisticsCaptor.capture(),
+ eq(true),
+ optionCaptor.capture());
+
+ assertThat(specCaptor.getAllValues()).extracting(List::size).containsExactly(1000, 1000, 1);
+ assertThat(statisticsCaptor.getAllValues())
+ .extracting(List::size)
+ .containsExactly(1000, 1000, 1);
+ assertThat(optionCaptor.getAllValues())
+ .extracting(List::size)
+ .containsExactly(1000, 1000, 1);
+ for (int batch = 0; batch < specCaptor.getAllValues().size(); batch++) {
+ List> batchSpecs = specCaptor.getAllValues().get(batch);
+ List batchStatistics = statisticsCaptor.getAllValues().get(batch);
+ List> batchOptions = optionCaptor.getAllValues().get(batch);
+ int offset = batch * REQUEST_SIZE;
+ assertThat(batchOptions)
+ .containsExactlyElementsOf(options.subList(offset, offset + batchSpecs.size()));
+ assertThat(batchStatistics)
+ .extracting(PartitionStatistics::spec)
+ .containsExactlyInAnyOrderElementsOf(batchSpecs);
+ }
+ }
+
+ @Test
+ void testInvalidPartitionOptionAfterBatchBoundaryTouchesNoCatalog() {
+ Catalog catalog = mock(Catalog.class);
+ List> specs = specs(1001);
+ List> options = new ArrayList<>(specs.size());
+ for (int i = 0; i < 1000; i++) {
+ options.add(Collections.emptyMap());
+ }
+ Map nullValue = new HashMap<>();
+ nullValue.put("path", null);
+ options.add(nullValue);
+
+ assertThatThrownBy(
+ () ->
+ partitionManager(catalog)
+ .createPartitions(
+ specs,
+ true,
+ Collections.singletonList(
+ statistics(specs.get(0), 0L)),
+ true,
+ options))
+ .isInstanceOf(IllegalArgumentException.class);
+ verifyNoInteractions(catalog);
+ }
+
@Test
void testMisalignedPartitionOptionsAreRejectedBeforeCatalogAccess() {
Catalog catalog = mock(Catalog.class);
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitEscapedPartitionValueTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitEscapedPartitionValueTest.java
new file mode 100644
index 000000000000..9fc12b1d2732
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitEscapedPartitionValueTest.java
@@ -0,0 +1,183 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.TwoPhaseOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.utils.PartitionPathUtils;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.ArgumentCaptor;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * A partition value is escaped into one directory name: {@code a%b} is written {@code a%25b}, and
+ * {@code a/b} is one level named {@code a%2Fb} rather than two. The directory an overwrite or a
+ * truncate names has to be that name, or the catalog cannot tell it is the partition's own.
+ */
+class FormatTableCommitEscapedPartitionValueTest {
+
+ private static final List VALUES = Arrays.asList("a%b", "a/b", "a=b");
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testOverwriteNamesTheEscapedDefaultDirectoryOfEveryPartition() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "escaped-overwrite");
+ List> specs = new ArrayList<>();
+ List messages = new ArrayList<>();
+ for (String value : VALUES) {
+ Map spec = Collections.singletonMap("part", value);
+ specs.add(spec);
+ Path directory = new Path(tablePath, partitionDirectory(spec));
+ fileIO.writeFile(new Path(directory, "data-old.csv"), "old", false);
+ TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class);
+ when(committer.targetPath()).thenReturn(new Path(directory, "data-new.csv"));
+ messages.add(new TwoPhaseCommitMessage(committer));
+ }
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+
+ commit(tablePath, fileIO, partitionManager, /* dynamicPartitionOverwrite */ true)
+ .commit(messages);
+
+ assertReportedDirectories(partitionManager, tablePath, specs);
+ }
+
+ @Test
+ void testTruncateNamesTheEscapedDefaultDirectoryAndLeavesExternalData() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "escaped-truncate");
+ Path externalRoot = new Path(new Path(tempDir.toUri()), "escaped-external");
+ List> specs = new ArrayList<>();
+ List registered = new ArrayList<>();
+ List externalData = new ArrayList<>();
+ for (int i = 0; i < VALUES.size(); i++) {
+ Map spec = Collections.singletonMap("part", VALUES.get(i));
+ specs.add(spec);
+ fileIO.writeFile(
+ new Path(new Path(tablePath, partitionDirectory(spec)), "data-old.csv"),
+ "old",
+ false);
+ Path external = new Path(new Path(externalRoot, "external-" + i), "data.csv");
+ fileIO.writeFile(external, "external", false);
+ externalData.add(external);
+ registered.add(partitionAt(spec, new Path(externalRoot, "external-" + i).toString()));
+ }
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ when(partitionManager.listPartitionsByNames(specs)).thenReturn(registered);
+
+ commit(tablePath, fileIO, partitionManager, /* dynamicPartitionOverwrite */ false)
+ .truncatePartitions(specs);
+
+ for (Path external : externalData) {
+ assertThat(fileIO.exists(external)).isTrue();
+ }
+ assertReportedDirectories(partitionManager, tablePath, specs);
+ }
+
+ private FormatTableCommit commit(
+ Path tablePath,
+ LocalFileIO fileIO,
+ FormatTablePartitionManager partitionManager,
+ boolean dynamicPartitionOverwrite) {
+ return new FormatTableCommit(
+ tablePath.toString(),
+ Collections.singletonList("part"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ dynamicPartitionOverwrite,
+ Identifier.create("escaped_db", "escaped_table"),
+ null,
+ null,
+ null,
+ partitionManager,
+ dynamicPartitionOverwrite);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static void assertReportedDirectories(
+ FormatTablePartitionManager partitionManager,
+ Path tablePath,
+ List> expectedSpecs) {
+ ArgumentCaptor>> specs =
+ ArgumentCaptor.forClass((Class) List.class);
+ ArgumentCaptor>> options =
+ ArgumentCaptor.forClass((Class) List.class);
+ verify(partitionManager)
+ .createPartitions(
+ specs.capture(),
+ eq(true),
+ org.mockito.ArgumentMatchers.anyList(),
+ eq(true),
+ options.capture());
+ assertThat(specs.getValue()).containsExactlyInAnyOrderElementsOf(expectedSpecs);
+ assertThat(options.getValue()).hasSameSizeAs(specs.getValue());
+ for (int i = 0; i < specs.getValue().size(); i++) {
+ Map spec = specs.getValue().get(i);
+ assertThat(options.getValue().get(i))
+ .as("the directory named for %s keeps the escapes of its value", spec)
+ .isEqualTo(
+ Collections.singletonMap(
+ CoreOptions.PATH.key(),
+ new Path(tablePath, partitionDirectory(spec)).toString()));
+ }
+ }
+
+ private static String partitionDirectory(Map spec) {
+ return PartitionPathUtils.generatePartitionPathUtil(new LinkedHashMap<>(spec), false);
+ }
+
+ private static Partition partitionAt(Map spec, String location) {
+ return new Partition(
+ spec,
+ 0,
+ 0,
+ 0,
+ 0,
+ PartitionStatistics.UNKNOWN_TOTAL_BUCKETS,
+ false,
+ null,
+ null,
+ null,
+ null,
+ Collections.singletonMap(CoreOptions.PATH.key(), location));
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitRegistryValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitRegistryValidationTest.java
index 7ffe000ddd49..675dad393945 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitRegistryValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitRegistryValidationTest.java
@@ -19,18 +19,22 @@
package org.apache.paimon.table.format;
import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.TwoPhaseOutputStream;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionStatistics;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.mockito.ArgumentCaptor;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
@@ -42,6 +46,7 @@
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -53,6 +58,193 @@ class FormatTableCommitRegistryValidationTest {
@TempDir java.nio.file.Path tempDir;
+ @Test
+ void testAppendValidatesOnlyAffectedRowsReturnedByName() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "append-scoped-registry");
+ Map targetSpec = partitionSpec("2025", "10");
+ Map unrelatedMalformedSpec = Collections.singletonMap("year", "2024");
+ TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class);
+ when(committer.targetPath())
+ .thenReturn(new Path(tablePath, "year=2025/month=10/data-new.csv"));
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ when(partitionManager.listPartitionsByNames(Collections.singletonList(targetSpec)))
+ .thenReturn(Collections.singletonList(partitionAt(targetSpec, null)));
+ when(partitionManager.listPartitions(Collections.emptyMap(), null))
+ .thenReturn(Collections.singletonList(partitionAt(unrelatedMalformedSpec, null)));
+ FormatTableCommit commit = commit(tablePath, fileIO, partitionManager, false, null, true);
+
+ commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer)));
+
+ verify(partitionManager).listPartitionsByNames(Collections.singletonList(targetSpec));
+ verify(partitionManager, never()).listPartitions(anyMap(), isNull());
+ verify(committer).commit(fileIO);
+ }
+
+ @Test
+ void testAppendRejectsMalformedAffectedRowBeforePublishing() throws Exception {
+ MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "append-malformed-affected-row");
+ Map targetSpec = partitionSpec("2025", "10");
+ Map incompleteSpec = Collections.singletonMap("year", "2025");
+ TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class);
+ when(committer.targetPath())
+ .thenReturn(new Path(tablePath, "year=2025/month=10/data-new.csv"));
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ when(partitionManager.listPartitionsByNames(Collections.singletonList(targetSpec)))
+ .thenReturn(Collections.singletonList(partitionAt(incompleteSpec, null)));
+ FormatTableCommit commit = commit(tablePath, fileIO, partitionManager, false, null, true);
+ fileIO.startTrackingMutations();
+
+ assertThatThrownBy(
+ () ->
+ commit.commit(
+ Collections.singletonList(
+ new TwoPhaseCommitMessage(committer))))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseMessage(
+ "Catalog returned incomplete partition spec {year=2025} for Format Table "
+ + "location_db.location_table.");
+
+ assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.mkdirsCalls()).isZero();
+ verify(committer, never()).commit(fileIO);
+ verify(committer, never()).clean(fileIO);
+ verify(partitionManager, never()).listPartitions(anyMap(), isNull());
+ verify(partitionManager, never())
+ .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), any());
+ }
+
+ @Test
+ void testStaticPrefixOverwriteValidatesOnlyRowsReturnedForPrefix() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "overwrite-scoped-prefix");
+ Map prefix = Collections.singletonMap("year", "2025");
+ Map targetSpec = partitionSpec("2025", "10");
+ Map unrelatedMalformedSpec = Collections.singletonMap("year", "2024");
+ Path targetData = new Path(tablePath, "year=2025/month=10/data-old.csv");
+ Path unrelatedData = new Path(tablePath, "year=2024/month=11/data-old.csv");
+ fileIO.writeFile(targetData, "target", false);
+ fileIO.writeFile(unrelatedData, "unrelated", false);
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ when(partitionManager.listPartitions(prefix, null))
+ .thenReturn(Collections.singletonList(partitionAt(targetSpec, null)));
+ when(partitionManager.listPartitions(Collections.emptyMap(), null))
+ .thenReturn(Collections.singletonList(partitionAt(unrelatedMalformedSpec, null)));
+ FormatTableCommit commit = commit(tablePath, fileIO, partitionManager, true, prefix, true);
+
+ commit.commit(Collections.emptyList());
+
+ assertThat(fileIO.exists(targetData)).isFalse();
+ assertThat(fileIO.exists(unrelatedData)).isTrue();
+ verify(partitionManager).listPartitions(prefix, null);
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ }
+
+ @Test
+ void testTruncatePrefixValidatesOnlyRowsReturnedForPrefix() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "truncate-scoped-prefix");
+ Map prefix = Collections.singletonMap("year", "2025");
+ Map targetSpec = partitionSpec("2025", "10");
+ Map unrelatedMalformedSpec = Collections.singletonMap("year", "2024");
+ Path targetData = new Path(tablePath, "year=2025/month=10/data-old.csv");
+ Path unrelatedData = new Path(tablePath, "year=2024/month=11/data-old.csv");
+ fileIO.writeFile(targetData, "target", false);
+ fileIO.writeFile(unrelatedData, "unrelated", false);
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ when(partitionManager.listPartitions(prefix, null))
+ .thenReturn(Collections.singletonList(partitionAt(targetSpec, null)));
+ when(partitionManager.listPartitions(Collections.emptyMap(), null))
+ .thenReturn(Collections.singletonList(partitionAt(unrelatedMalformedSpec, null)));
+ FormatTableCommit commit = commit(tablePath, fileIO, partitionManager, false, null, true);
+
+ commit.truncatePartitions(Collections.singletonList(prefix));
+
+ assertThat(fileIO.exists(targetData)).isFalse();
+ assertThat(fileIO.exists(unrelatedData)).isTrue();
+ verify(partitionManager).listPartitions(prefix, null);
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ }
+
+ @Test
+ void testTruncateMixedExactAndPrefixTargetsUsesScopedLookupsAndDeduplicates() throws Exception {
+ MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "truncate-mixed-targets");
+ Map exactOutsidePrefix = partitionSpec("2024", "12");
+ Map exactInsidePrefix = partitionSpec("2025", "10");
+ Map prefixOnly = partitionSpec("2025", "11");
+ Map prefix = Collections.singletonMap("year", "2025");
+ Path exactOutsideData = new Path(tablePath, "year=2024/month=12/data-exact-outside.csv");
+ Path exactInsideData = new Path(tablePath, "year=2025/month=10/data-exact-inside.csv");
+ Path prefixOnlyData = new Path(tablePath, "year=2025/month=11/data-prefix-only.csv");
+ Path unrelatedData = new Path(tablePath, "year=2023/month=01/data-unrelated.csv");
+ fileIO.writeFile(exactOutsideData, "outside", false);
+ fileIO.writeFile(exactInsideData, "inside", false);
+ fileIO.writeFile(prefixOnlyData, "prefix", false);
+ fileIO.writeFile(unrelatedData, "unrelated", false);
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ List> exactSpecs = Arrays.asList(exactOutsidePrefix, exactInsidePrefix);
+ Partition exactInsidePartition = partitionAt(exactInsidePrefix, null);
+ when(partitionManager.listPartitionsByNames(exactSpecs))
+ .thenReturn(
+ Arrays.asList(partitionAt(exactOutsidePrefix, null), exactInsidePartition));
+ when(partitionManager.listPartitions(prefix, null))
+ .thenReturn(Arrays.asList(exactInsidePartition, partitionAt(prefixOnly, null)));
+ FormatTableCommit commit = commit(tablePath, fileIO, partitionManager, false, null, true);
+ fileIO.startTrackingMutations();
+
+ commit.truncatePartitions(Arrays.asList(exactOutsidePrefix, exactInsidePrefix, prefix));
+
+ assertThat(fileIO.exists(exactOutsideData)).isFalse();
+ assertThat(fileIO.exists(exactInsideData)).isFalse();
+ assertThat(fileIO.exists(prefixOnlyData)).isFalse();
+ assertThat(fileIO.exists(unrelatedData)).isTrue();
+ assertThat(fileIO.deleteCalls()).isEqualTo(3);
+ assertThat(fileIO.listCalls(exactInsideData.getParent())).isEqualTo(1);
+ assertThat(fileIO.mkdirsCalls()).isZero();
+ verify(partitionManager).listPartitionsByNames(exactSpecs);
+ verify(partitionManager).listPartitions(prefix, null);
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ assertReplacementReport(
+ partitionManager,
+ tablePath,
+ Arrays.asList(exactOutsidePrefix, exactInsidePrefix, prefixOnly));
+ }
+
+ @Test
+ void testWholeTableTruncateValidatesFullRegistryBeforeMutation() throws Exception {
+ MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
+ Path tablePath = new Path(new Path(tempDir.toUri()), "truncate-full-registry");
+ Map validSpec = partitionSpec("2025", "10");
+ Map