You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/09/10 08:53:37 UTC
(paimon) branch master updated: [api][core] Reset Format Table partition locations on overwrite (#9710)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new bc73d8953e [api][core] Reset Format Table partition locations on overwrite (#9710)
bc73d8953e is described below
commit bc73d8953e9031cd5a08aa1fba2a7ffc1f9d80c3
Author: Dapeng Sun(孙大鹏) <da...@alibaba-inc.com>
AuthorDate: Thu Sep 10 16:53:29 2026 +0800
[api][core] Reset Format Table partition locations on overwrite (#9710)
---
docs/docs/concepts/rest/rest-api.md | 7 +
docs/scripts/validate-rest-openapi.js | 7 +-
docs/static/rest-catalog-open-api.yaml | 4 +-
.../main/java/org/apache/paimon/rest/RESTApi.java | 8 +-
.../rest/requests/CreatePartitionsRequest.java | 9 +-
.../java/org/apache/paimon/catalog/Catalog.java | 5 +
.../java/org/apache/paimon/rest/RESTCatalog.java | 6 +-
.../format/CatalogFormatTablePartitionManager.java | 2 +-
.../paimon/table/format/FormatTableCommit.java | 391 ++++++++-------
.../table/format/FormatTablePartitionManager.java | 7 +-
.../format/FormatTablePartitionPathResolver.java | 85 +++-
.../apache/paimon/rest/MockRESTCatalogTest.java | 407 +++++++++++++++-
.../org/apache/paimon/rest/RESTApiJsonTest.java | 43 +-
.../paimon/rest/RESTCatalogPartitionSupport.java | 229 ++++++++-
.../org/apache/paimon/rest/RESTCatalogServer.java | 22 +-
.../CatalogFormatTablePartitionManagerTest.java | 86 ++++
...FormatTableCommitEscapedPartitionValueTest.java | 183 +++++++
.../FormatTableCommitRegistryValidationTest.java | 270 +++++++++++
.../format/FormatTableCommitStatisticsTest.java | 231 ++++++---
.../paimon/table/format/FormatTableCommitTest.java | 523 +++++++++++----------
.../FormatTablePartitionPathResolverTest.java | 51 ++
21 files changed, 2021 insertions(+), 555 deletions(-)
diff --git a/docs/docs/concepts/rest/rest-api.md b/docs/docs/concepts/rest/rest-api.md
index d7c2114058..49351779ba 100644
--- a/docs/docs/concepts/rest/rest-api.md
+++ b/docs/docs/concepts/rest/rest-api.md
@@ -30,6 +30,13 @@ order of `partitionSpecs`; use `{}` when a partition has no options. Custom loca
`path` option. Before registering custom locations, ensure that the REST server supports partition
options and all readers support custom locations.
+For an existing Format Table partition, omitting `path` keeps its location. Naming the partition's
+own default directory under the table asks the server to put it back there: the stored location is
+dropped, no data is deleted, and the request needs `replaceStatistics=true` with a
+`partitionStatistics` entry for the same spec. Any other path under the table location stays
+invalid, so a server that does not implement this rejects the request rather than storing it. A
+server also rejects additive statistics for a partition that already has a custom location.
+
<body>
<iframe src="/docs/master/rest-catalog-open-api.yaml" width="100%" height="800px" />
</body>
diff --git a/docs/scripts/validate-rest-openapi.js b/docs/scripts/validate-rest-openapi.js
index 100dcb8c66..3dd73928aa 100644
--- a/docs/scripts/validate-rest-openapi.js
+++ b/docs/scripts/validate-rest-openapi.js
@@ -275,8 +275,11 @@ function validateCatalogOpenApi() {
requestOptions.description.includes('partitionSpecs') &&
requestOptions.description.toLowerCase().includes('position') &&
requestOptions.description.toLowerCase().includes('same length') &&
- requestOptions.description.toLowerCase().includes('empty object'),
- 'CreatePartitionsRequest.partitionOptions must document positional alignment and empty options',
+ requestOptions.description.toLowerCase().includes('empty object') &&
+ requestOptions.description.includes('replaceStatistics=true') &&
+ requestOptions.description.includes('partitionStatistics') &&
+ requestOptions.description.toLowerCase().includes('default directory'),
+ 'CreatePartitionsRequest.partitionOptions must document alignment and returns to the default directory',
);
contract.requireProperties('Partition', ['options']);
contract.requireProperties('ConfigResponse', ['defaults', 'overrides']);
diff --git a/docs/static/rest-catalog-open-api.yaml b/docs/static/rest-catalog-open-api.yaml
index f3c096b45a..772555f067 100644
--- a/docs/static/rest-catalog-open-api.yaml
+++ b/docs/static/rest-catalog-open-api.yaml
@@ -2506,10 +2506,10 @@ components:
items:
$ref: '#/components/schemas/PartitionStatistics'
replaceStatistics:
- description: Whether partitionStatistics replace the stored values rather than add to them; required whenever partitionStatistics is present, and absent otherwise. Replacing overwrites recordCount, fileSizeInBytes, fileCount and lastFileCreationTime; adding sums the three counts and keeps the later lastFileCreationTime, since two timestamps do not add. A field reported as unknown leaves the stored one alone either way, and totalBuckets is never combined. A client that reports o [...]
+ description: Whether partitionStatistics replace the stored values rather than add to them; required whenever partitionStatistics is present, and absent otherwise. Replacing overwrites recordCount, fileSizeInBytes, fileCount and lastFileCreationTime; adding sums the three counts and keeps the later lastFileCreationTime, since two timestamps do not add. A field reported as unknown leaves the stored one alone either way, and totalBuckets is never combined. A client that reports o [...]
type: [ boolean, "null" ]
partitionOptions:
- description: Optional options aligned with partitionSpecs by position. The list must have the same length as partitionSpecs; use an empty object when a partition has no options. A custom location is stored under the path key.
+ description: Optional options aligned with partitionSpecs by position. The list must have the same length as partitionSpecs; use an empty object when a partition has no options. A custom location is stored under the path key. Omitting path keeps the stored location; a path outside the table location registers a custom location for a new partition and must equal the stored path for an existing one; the partition's own default directory under the table location returns the partit [...]
type: [ array, "null" ]
items:
type: object
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
index 84cdecba48..6dbe7b872e 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -1019,6 +1019,11 @@ public class RESTApi {
* as unknown leaves the stored one alone either way, and a report never creates or removes a
* partition row.
*
+ * <p>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 class RESTApi {
* 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 4e0ec7ab82..e5260a8dcb 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 @@ import static org.apache.paimon.utils.Preconditions.checkArgument;
*
* <p>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 class CreatePartitionsRequest implements RESTRequest {
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<Map<String, String>> 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 329abcf67b..7e824845db 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 @@ public interface Catalog extends AutoCloseable {
/**
* Create partitions atomically unless existing entries are ignored, with optional statistics
* and position-aligned options.
+ *
+ * <p>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 0fb356966d..71cdb10123 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 @@ public class RESTCatalog implements Catalog {
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 449ef83ab5..72e23a200c 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 @@ class CatalogFormatTablePartitionManager implements FormatTablePartitionManager
@Nullable List<PartitionStatistics> statistics,
boolean replaceStatistics,
@Nullable List<Map<String, String>> partitionOptions) {
- validatePartitionOptions(partitionOptions, partitions);
Map<Map<String, String>, 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 d7169cfdd5..b823fc462e 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.ExecutorService;
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 class FormatTableCommit implements BatchTableCommit {
}
}
- List<Partition> validatedPartitions = rejectWritesToCustomLocationPartitions(messages);
+ Set<Map<String, String>> writtenPartitionSpecs;
+ try {
+ writtenPartitionSpecs = writtenPartitionSpecs(messages);
+ } catch (RuntimeException failure) {
+ markPublishedTargetsToPreserveOnAbort(messages);
+ throw failure;
+ }
+ List<Partition> targetPartitions =
+ prepareCatalogManagedCommit(messages, writtenPartitionSpecs);
- Set<Map<String, String>> partitionSpecs = new HashSet<>();
- Set<Path> clearedPartitionPaths = new HashSet<>();
+ Set<Map<String, String>> partitionSpecs = new LinkedHashSet<>();
+ Set<Map<String, String>> 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 class FormatTableCommit implements BatchTableCommit {
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 class FormatTableCommit implements BatchTableCommit {
// 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 class FormatTableCommit implements BatchTableCommit {
extractPartitionSpecFromPath(
committer.targetPath().getParent(), partitionKeys);
partitionSpecs.add(spec);
+ reportTargetSpecs.add(spec);
if (reportsStatistics) {
statisticsByPartition.merge(
spec,
@@ -298,12 +318,7 @@ public class FormatTableCommit implements BatchTableCommit {
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 class FormatTableCommit implements BatchTableCommit {
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 @@ public class FormatTableCommit implements BatchTableCommit {
}
}
- /** Rejects writes whose files would belong to a catalog partition outside the table root. */
- private List<Partition> rejectWritesToCustomLocationPartitions(
- List<TwoPhaseCommitMessage> messages) {
+ /** Loads the registry rows this operation needs before any table mutation. */
+ private List<Partition> prepareCatalogManagedCommit(
+ List<TwoPhaseCommitMessage> messages, Set<Map<String, String>> writtenPartitionSpecs) {
if (partitionManager == null || partitionKeys == null || partitionKeys.isEmpty()) {
return Collections.emptyList();
}
try {
- return rejectWritesToCustomLocationPartitionsBeforeMutation(messages);
+ List<Partition> 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 @@ public class FormatTableCommit implements BatchTableCommit {
}
}
- private List<Partition> rejectWritesToCustomLocationPartitionsBeforeMutation(
- List<TwoPhaseCommitMessage> messages) {
- Predicate<Partition> affectsPartition;
- boolean hasStaticPrefixWithoutFiles = false;
- if (overwrite && staticPartitions != null && !staticPartitions.isEmpty()) {
- LinkedHashMap<String, String> 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<Map<String, String>> 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<Partition> loadCommitTargetPartitions(
+ Set<Map<String, String>> writtenPartitionSpecs) {
+ if (overwrite) {
+ if (staticPartitions == null || staticPartitions.isEmpty()) {
+ return replacesOnlyWrittenPartitions()
+ ? Collections.emptyList()
+ : loadPartitionRegistry();
}
- if (!overwrite && staticPartitions != null && !staticPartitions.isEmpty()) {
- LinkedHashMap<String, String> 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<Partition> registry = loadPartitionRegistry();
- List<Partition> affectedPartitions =
- registry.stream().filter(affectsPartition).collect(Collectors.toList());
- for (Partition partition : affectedPartitions) {
- if (FormatTablePartitionPathResolver.customLocation(partition) != null) {
- throw unsupportedCustomLocation(overwrite ? "Overwriting" : "Writing", partition);
+ LinkedHashMap<String, String> staticSpec = orderedPartitionPrefix(staticPartitions);
+ return staticSpec.size() == partitionKeys.size()
+ ? Collections.emptyList()
+ : loadPartitionsByPrefix(staticSpec);
+ }
+
+ return writtenPartitionSpecs.isEmpty()
+ ? Collections.emptyList()
+ : loadPartitionsByNames(writtenPartitionSpecs);
+ }
+
+ private Set<Map<String, String>> writtenPartitionSpecs(List<TwoPhaseCommitMessage> messages) {
+ if (partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptySet();
+ }
+ Set<Map<String, String>> 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<String, String> orderedPartitionPrefix(
@@ -492,9 +491,42 @@ public class FormatTableCommit implements BatchTableCommit {
return orderedSpec;
}
- /** Validates every registered path before using it for a write or truncate decision. */
+ private List<Partition> loadPartitionsByNames(Set<Map<String, String>> partitionSpecs) {
+ List<Partition> 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<Partition> loadPartitionsByPrefix(Map<String, String> prefix) {
+ List<Partition> 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<Partition> loadPartitionRegistry() {
- List<Partition> partitions = partitionManager.listPartitions(Collections.emptyMap(), null);
+ return validatePartitionRegistry(
+ partitionManager.listPartitions(Collections.emptyMap(), null));
+ }
+
+ private List<Partition> validatePartitionRegistry(List<Partition> partitions) {
FormatTablePartitionRegistryValidator.validatePartitionLocations(
partitions,
partitionKeys,
@@ -544,39 +576,46 @@ public class FormatTableCommit implements BatchTableCommit {
}
/**
- * 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<Map<String, String>> writtenPartitionSpecs,
+ Set<Map<String, String>> targetPartitionSpecs,
Map<Map<String, String>, PartitionStatistics> statisticsByPartition,
- Set<Path> clearedPartitionPaths,
long commitTime,
boolean replaceStatistics) {
- for (Path cleared : clearedPartitionPaths) {
- Map<String, String> 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<Map<String, String>> specs = new LinkedHashSet<>(writtenPartitionSpecs);
+ Set<Map<String, String>> specs = new LinkedHashSet<>(targetPartitionSpecs);
specs.addAll(statisticsByPartition.keySet());
if (specs.isEmpty()) {
return;
}
+
+ List<Map<String, String>> 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<String, String> 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 @@ public class FormatTableCommit implements BatchTableCommit {
: 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<String, String> clearedPartitionSpec(Path clearedPath) {
- LinkedHashMap<String, String> 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 @@ public class FormatTableCommit implements BatchTableCommit {
/**
* 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<Path> tableDataDirectories(List<Partition> validatedPartitions) {
+ private List<Path> tableDataDirectories(
+ List<Partition> targetPartitions, Set<Map<String, String>> writtenPartitionSpecs) {
if (partitionKeys == null || partitionKeys.isEmpty()) {
return Collections.singletonList(new Path(location));
}
List<Path> directories = new ArrayList<>();
if (partitionManager != null) {
- for (Partition partition : validatedPartitions) {
+ Set<Map<String, String>> targetSpecs = new LinkedHashSet<>();
+ for (Partition partition : targetPartitions) {
+ targetSpecs.add(partition.spec());
+ }
+ targetSpecs.addAll(writtenPartitionSpecs);
+ for (Map<String, String> targetSpec : targetSpecs) {
directories.add(
buildPartitionPath(
location,
- partition.spec(),
+ targetSpec,
formatTablePartitionOnlyValueInPath,
partitionKeys));
}
@@ -1211,11 +1202,6 @@ public class FormatTableCommit implements BatchTableCommit {
// by whatever the table reads its partitions from.
if (partitionManager != null) {
List<Partition> 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 class FormatTableCommit implements BatchTableCommit {
truncate(normalizedSpecs);
return;
}
- List<Partition> registry = loadPartitionRegistry();
- Map<Map<String, String>, Partition> partitions =
- selectRequestedPartitions(registry, normalizedSpecs);
- for (Partition partition : partitions.values()) {
- if (FormatTablePartitionPathResolver.customLocation(partition) != null) {
- throw unsupportedCustomLocation("Truncating", partition);
+ List<Partition> partitions = loadRequestedPartitions(normalizedSpecs);
+ truncate(partitions.stream().map(Partition::spec).collect(Collectors.toList()));
+ }
+
+ private List<Partition> loadRequestedPartitions(List<Map<String, String>> partitionSpecs) {
+ Set<Map<String, String>> exactSpecs = new LinkedHashSet<>();
+ Set<Map<String, String>> prefixes = new LinkedHashSet<>();
+ for (Map<String, String> 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<Partition> found = new ArrayList<>();
+ if (!exactSpecs.isEmpty()) {
+ found.addAll(loadPartitionsByNames(exactSpecs));
+ }
+ for (Map<String, String> prefix : prefixes) {
+ found.addAll(loadPartitionsByPrefix(prefix));
+ }
- private Map<Map<String, String>, Partition> selectRequestedPartitions(
- List<Partition> registry, List<Map<String, String>> partitionSpecs) {
Map<Map<String, String>, Partition> selected = new LinkedHashMap<>();
- Set<Map<String, String>> requestedPrefixes = new HashSet<>(partitionSpecs);
- for (Partition partition : registry) {
- if (requestedPrefixes.contains(Collections.emptyMap())) {
- selected.putIfAbsent(partition.spec(), partition);
- continue;
- }
- LinkedHashMap<String, String> 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<Map<String, String>> partitionSpecs) {
long truncateTime = System.currentTimeMillis();
- Set<Path> 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<Map<String, String>, PartitionStatistics> emptied = new LinkedHashMap<>();
@@ -1295,9 +1290,7 @@ public class FormatTableCommit implements BatchTableCommit {
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 @@ public class FormatTableCommit implements BatchTableCommit {
// 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 0c74250e41..aebbcfd1fb 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 @@ public interface FormatTablePartitionManager extends Serializable {
* 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.
*
* <p>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 6c5c6327e2..620c1b2690 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 @@ public final class FormatTablePartitionPathResolver {
}
Path resolve(LinkedHashMap<String, String> 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 @@ public final class FormatTablePartitionPathResolver {
}
}
+ /** Where a partition lives when it carries no location of its own. */
+ public static Path defaultPartitionPath(
+ Path tablePath, LinkedHashMap<String, String> 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<String, String> 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 @@ public final class FormatTablePartitionPathResolver {
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 final class FormatTablePartitionPathResolver {
}
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 210db841eb..492b509dfd 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 @@ class MockRESTCatalogTest extends RESTCatalogTest {
}
@Test
- void testInvalidCustomPartitionLocationFailsBeforePost() throws Exception {
+ void testInvalidCustomPartitionLocationIsRejectedWithoutMutation() throws Exception {
Identifier identifier = createFormatTableWithCatalogManagedPartitions();
Map<String, String> spec = Collections.singletonMap("dt", "20260717");
String partitionsResource =
@@ -609,10 +609,12 @@ class MockRESTCatalogTest extends RESTCatalogTest {
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 @@ class MockRESTCatalogTest extends RESTCatalogTest {
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 @@ class MockRESTCatalogTest extends RESTCatalogTest {
.containsExactly(tuple(spec, originalLocation));
}
+ @Test
+ void testExistingPartitionAcceptsTheSameCustomLocationIdempotently() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map<String, String> spec = Collections.singletonMap("dt", "20260717");
+ Map<String, String> 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<String, String> spec = Collections.singletonMap("dt", "20260717");
+ Map<String, String> 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<String, String> spec = Collections.singletonMap("dt", "20260717");
+ Map<String, String> newSpec = Collections.singletonMap("dt", "20260718");
+ String releasedLocation = "file:/archive/original";
+ Map<String, String> 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<Map<String, String>> specs = Arrays.asList(spec, newSpec);
+ List<PartitionStatistics> replacement =
+ Arrays.asList(partitionStatistics(spec, 3L), partitionStatistics(newSpec, 7L));
+ Map<String, String> resetOptions = returnToDefault(identifier, spec);
+ resetOptions.put("owner", "must-not-overwrite-existing-options");
+ List<Map<String, String>> 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<String, String> ordinarySpec = Collections.singletonMap("dt", "20260717");
+ Map<String, String> 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<String, String> ordinary = Collections.singletonMap("dt", "20260717");
+ Map<String, String> custom = Collections.singletonMap("dt", "20260718");
+ Map<String, String> 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<Partition> 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<String, String> resetSpec = Collections.singletonMap("dt", "20260717");
+ Map<String, String> stableSpec = Collections.singletonMap("dt", "20260718");
+ Map<String, String> newSpec = Collections.singletonMap("dt", "20260719");
+ Map<String, String> 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<Partition> 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<String, String> resetSpec = Collections.singletonMap("dt", "20260717");
+ Map<String, String> otherSpec = Collections.singletonMap("dt", "20260718");
+ Map<String, String> 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<Partition> 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<String, String> resetSpec = Collections.singletonMap("dt", "20260717");
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(resetSpec),
+ true,
+ Collections.singletonList(partitionStatistics(resetSpec, 10L)),
+ true,
+ partitionOptions("file:/archive/reset-target"));
+ List<Partition> 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<String, String> spec = Collections.singletonMap("dt", "20260717");
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ Collections.singletonList(partitionStatistics(spec, 10L)),
+ true,
+ partitionOptions("file:/archive/reset-target"));
+ List<Partition> 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 @@ class MockRESTCatalogTest extends RESTCatalogTest {
Identifier identifier = createFormatTableWithCatalogManagedPartitions();
Map<String, String> spec = Collections.singletonMap("dt", "20260717");
List<Map<String, String>> specs = Collections.singletonList(spec);
- String location = "file:/archive/dt=20260717";
- Map<String, String> options = new HashMap<>();
- options.put(CoreOptions.PATH.key(), location);
- options.put("owner", "data-platform");
+ Map<String, String> options = Collections.singletonMap("owner", "data-platform");
FormatTablePartitionManager partitionManager =
((FormatTable) restCatalog.getTable(identifier)).partitionManager();
assertThat(partitionManager).isNotNull();
@@ -1058,6 +1400,33 @@ class MockRESTCatalogTest extends RESTCatalogTest {
return options;
}
+ /** The request that asks for a partition to live at its own default directory again. */
+ private Map<String, String> returnToDefault(Identifier identifier, Map<String, String> spec)
+ throws Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.PATH.key(), defaultPartitionDirectory(identifier, spec));
+ return options;
+ }
+
+ private String defaultPartitionDirectory(Identifier identifier, Map<String, String> 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<String, String> 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 @@ class MockRESTCatalogTest extends RESTCatalogTest {
private static class RawCreatePartitionsRequest implements RESTRequest {
private final List<Map<String, String>> partitionSpecs;
+ private final List<PartitionStatistics> partitionStatistics;
+ private final Boolean replaceStatistics;
private final List<Map<String, String>> partitionOptions;
private RawCreatePartitionsRequest(
List<Map<String, String>> partitionSpecs,
List<Map<String, String>> partitionOptions) {
+ this(partitionSpecs, null, null, partitionOptions);
+ }
+
+ private RawCreatePartitionsRequest(
+ List<Map<String, String>> partitionSpecs,
+ List<PartitionStatistics> partitionStatistics,
+ Boolean replaceStatistics,
+ List<Map<String, String>> partitionOptions) {
this.partitionSpecs = partitionSpecs;
+ this.partitionStatistics = partitionStatistics;
+ this.replaceStatistics = replaceStatistics;
this.partitionOptions = partitionOptions;
}
@@ -1492,6 +1873,16 @@ class MockRESTCatalogTest extends RESTCatalogTest {
return partitionOptions;
}
+ @JsonGetter("partitionStatistics")
+ public List<PartitionStatistics> 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 9641cf02f0..e16233ece5 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 class RESTApiJsonTest {
}
@Test
- public void createPartitionsRequestRejectsNullOptionMapsTest() {
+ public void createPartitionsRequestCarriesPartitionLocationsTest() throws Exception {
+ Map<String, String> returningSpec = Collections.singletonMap("dt", "20260901");
+ Map<String, String> untouchedSpec = Collections.singletonMap("dt", "20260902");
+ Map<String, String> 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<Map<String, String>> specs =
Arrays.asList(
Collections.singletonMap("dt", "20260901"),
@@ -377,6 +411,7 @@ public class RESTApiJsonTest {
Arrays.asList(null, Collections.emptyMap())));
Map<String, String> nullValue = new HashMap<>();
+ nullValue.put("path", null);
nullValue.put("owner", null);
assertThrows(
IllegalArgumentException.class,
@@ -384,8 +419,10 @@ public class RESTApiJsonTest {
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<String, String> 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 a7036a9ee9..28de8abd30 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 javax.annotation.Nullable;
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 @@ final class RESTCatalogPartitionSupport {
Map<String, String> 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 @@ final class RESTCatalogPartitionSupport {
static Partition newPartition(Map<String, String> spec, @Nullable Map<String, String> options) {
return new Partition(
- spec,
+ new LinkedHashMap<>(spec),
PartitionStatistics.UNKNOWN,
PartitionStatistics.UNKNOWN,
PartitionStatistics.UNKNOWN,
@@ -152,10 +157,228 @@ final class RESTCatalogPartitionSupport {
null,
null,
null,
+ normalizeNewPartitionOptions(options));
+ }
+
+ static List<Partition> copyPartitions(List<Partition> partitions) {
+ List<Partition> copied = new ArrayList<>(partitions.size());
+ for (Partition partition : partitions) {
+ copied.add(copyPartition(partition, copyOptions(partition.options())));
+ }
+ return copied;
+ }
+
+ private static void validateNoAdditiveStatisticsForCustomPartitions(
+ List<Partition> stored,
+ @Nullable List<PartitionStatistics> statistics,
+ @Nullable Boolean replaceStatistics) {
+ if (statistics == null || statistics.isEmpty() || Boolean.TRUE.equals(replaceStatistics)) {
+ return;
+ }
+ Set<Map<String, String>> 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<Map<String, String>> takeReturnsToDefault(
+ CreatePartitionsRequest request,
+ @Nullable List<Map<String, String>> requestedOptions,
+ List<Partition> 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<Map<String, String>> 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<Map<String, String>> requestedOptions, CatalogContext catalogContext) {
+ if (requestedOptions == null) {
+ return;
+ }
+ for (Map<String, String> options : requestedOptions) {
+ String location = options.get(PATH.key());
+ if (location != null) {
+ options.put(
+ PATH.key(),
+ FormatTablePartitionPathResolver.canonicalizeCustomLocation(
+ location, catalogContext)
+ .toString());
+ }
+ }
+ }
+
+ private static Set<Map<String, String>> defaultDirectoryRequests(
+ List<Map<String, String>> requestedSpecs,
+ @Nullable List<Map<String, String>> 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<String> partitionKeys = metadata.schema().partitionKeys();
+ boolean onlyValueInPath =
+ new CoreOptions(metadata.schema().options()).formatTablePartitionOnlyValueInPath();
+ Set<Map<String, String>> returning = new HashSet<>();
+ for (int i = 0; i < requestedOptions.size(); i++) {
+ Map<String, String> options = requestedOptions.get(i);
+ String requested = options.get(PATH.key());
+ Map<String, String> spec = requestedSpecs.get(i);
+ if (requested == null || spec == null || !spec.keySet().containsAll(partitionKeys)) {
+ continue;
+ }
+ LinkedHashMap<String, String> 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<Map<String, String>> returning,
+ @Nullable List<PartitionStatistics> statistics,
+ @Nullable Boolean replaceStatistics) {
+ if (returning.isEmpty()) {
+ return;
+ }
+ Set<Map<String, String>> reportedSpecs = new HashSet<>();
+ if (statistics != null) {
+ for (PartitionStatistics statistic : statistics) {
+ reportedSpecs.add(statistic.spec());
+ }
+ }
+ for (Map<String, String> 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<Partition> partitions,
+ Set<Map<String, String>> returningToDefault,
+ boolean formatTable,
+ TableMetadata metadata,
+ String tableName,
+ CatalogContext catalogContext) {
+ applyReturnsToDefault(partitions, returningToDefault);
+ if (formatTable) {
+ validateFormatTablePartitionLocations(partitions, metadata, tableName, catalogContext);
+ }
+ }
+
+ private static void applyReturnsToDefault(
+ List<Partition> partitions, Set<Map<String, String>> 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<String, String> normalizeNewPartitionOptions(
+ @Nullable Map<String, String> options) {
+ Map<String, String> copied = copyOptions(options);
+ if (copied == null) {
+ return null;
+ }
+ return copied.isEmpty() ? null : copied;
+ }
+
+ @Nullable
+ private static Map<String, String> withoutPath(@Nullable Map<String, String> options) {
+ Map<String, String> copied = copyOptions(options);
+ if (copied == null) {
+ return null;
+ }
+ copied.remove(PATH.key());
+ return copied.isEmpty() ? null : copied;
+ }
+
+ @Nullable
+ private static Map<String, String> copyOptions(@Nullable Map<String, String> options) {
+ return options == null ? null : new HashMap<>(options);
+ }
+
+ private static Partition copyPartition(
+ Partition partition, @Nullable Map<String, String> 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<Partition> 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 90a948395b..d196e3be23 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 @@ public class RESTCatalogServer {
throw new Catalog.TableNotExistException(tableIdentifier);
}
List<Partition> storedPartitions =
- new ArrayList<>(
+ RESTCatalogPartitionSupport.copyPartitions(
tablePartitionsStore.getOrDefault(
tableName, Collections.emptyList()));
Set<Map<String, String>> existingSpecs =
@@ -2237,6 +2237,15 @@ public class RESTCatalogServer {
return mockResponse(response, 409);
}
}
+ boolean formatTable = isFormatTable(tableMetadata.schema().toSchema());
+ Set<Map<String, String>> returningToDefault =
+ RESTCatalogPartitionSupport.takeReturnsToDefault(
+ request,
+ requestedOptions,
+ storedPartitions,
+ tableMetadata,
+ tableName,
+ catalogContext);
Optional<Map<String, String>> conflictingLocation =
RESTCatalogPartitionSupport.conflictingLocation(
storedPartitions, request.getPartitionSpecs(), requestedOptions);
@@ -2261,10 +2270,13 @@ public class RESTCatalogServer {
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 3c51dbdc64..d313c01d5c 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 org.mockito.ArgumentCaptor;
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 @@ class CatalogFormatTablePartitionManagerTest {
assertThat(flatten(optionCaptor.getAllValues())).isEqualTo(options);
}
+ @Test
+ void testPartitionLocationsStayWithReplacementStatisticsAcrossBatches() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ List<Map<String, String>> specs = specs(2001);
+ List<Map<String, String>> options = new ArrayList<>(specs.size());
+ List<PartitionStatistics> 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<List<Map<String, String>>> specCaptor = ArgumentCaptor.forClass(List.class);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<PartitionStatistics>> statisticsCaptor =
+ ArgumentCaptor.forClass(List.class);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<List<Map<String, String>>> 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<Map<String, String>> batchSpecs = specCaptor.getAllValues().get(batch);
+ List<PartitionStatistics> batchStatistics = statisticsCaptor.getAllValues().get(batch);
+ List<Map<String, String>> 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<Map<String, String>> specs = specs(1001);
+ List<Map<String, String>> options = new ArrayList<>(specs.size());
+ for (int i = 0; i < 1000; i++) {
+ options.add(Collections.emptyMap());
+ }
+ Map<String, String> 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 0000000000..9fc12b1d27
--- /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<String> 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<Map<String, String>> specs = new ArrayList<>();
+ List<CommitMessage> messages = new ArrayList<>();
+ for (String value : VALUES) {
+ Map<String, String> 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<Map<String, String>> specs = new ArrayList<>();
+ List<Partition> registered = new ArrayList<>();
+ List<Path> externalData = new ArrayList<>();
+ for (int i = 0; i < VALUES.size(); i++) {
+ Map<String, String> 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<Map<String, String>> expectedSpecs) {
+ ArgumentCaptor<List<Map<String, String>>> specs =
+ ArgumentCaptor.forClass((Class) List.class);
+ ArgumentCaptor<List<Map<String, String>>> 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<String, String> 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<String, String> spec) {
+ return PartitionPathUtils.generatePartitionPathUtil(new LinkedHashMap<>(spec), false);
+ }
+
+ private static Partition partitionAt(Map<String, String> 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 7ffe000ddd..675dad3939 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.any;
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<String, String> targetSpec = partitionSpec("2025", "10");
+ Map<String, String> 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<String, String> targetSpec = partitionSpec("2025", "10");
+ Map<String, String> 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<String, String> prefix = Collections.singletonMap("year", "2025");
+ Map<String, String> targetSpec = partitionSpec("2025", "10");
+ Map<String, String> 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<String, String> prefix = Collections.singletonMap("year", "2025");
+ Map<String, String> targetSpec = partitionSpec("2025", "10");
+ Map<String, String> 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<String, String> exactOutsidePrefix = partitionSpec("2024", "12");
+ Map<String, String> exactInsidePrefix = partitionSpec("2025", "10");
+ Map<String, String> prefixOnly = partitionSpec("2025", "11");
+ Map<String, String> 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<Map<String, String>> 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<String, String> validSpec = partitionSpec("2025", "10");
+ Map<String, String> incompleteSpec = Collections.singletonMap("year", "2024");
+ Path validData = new Path(tablePath, "year=2025/month=10/data-old.csv");
+ fileIO.writeFile(validData, "old", false);
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ when(partitionManager.listPartitions(Collections.emptyMap(), null))
+ .thenReturn(
+ Arrays.asList(
+ partitionAt(validSpec, null), partitionAt(incompleteSpec, null)));
+ FormatTableCommit commit = commit(tablePath, fileIO, partitionManager, false, null, true);
+ fileIO.startTrackingMutations();
+
+ assertThatThrownBy(commit::truncateTable)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Catalog returned incomplete partition spec {year=2024} for Format Table "
+ + "location_db.location_table.");
+
+ assertThat(fileIO.exists(validData)).isTrue();
+ assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.mkdirsCalls()).isZero();
+ verify(partitionManager).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ verify(partitionManager, never())
+ .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), any());
+ }
+
@Test
void testTruncateRejectsNonLeadingPrefixBeforeMutationWhenFullRegistryIsRequired()
throws Exception {
@@ -122,6 +314,28 @@ class FormatTableCommitRegistryValidationTest {
options);
}
+ private static FormatTableCommit commit(
+ Path tablePath,
+ LocalFileIO fileIO,
+ FormatTablePartitionManager partitionManager,
+ boolean overwrite,
+ Map<String, String> staticPartitions,
+ boolean dynamicPartitionOverwrite) {
+ return new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ false,
+ PARTITION_DEFAULT_NAME.defaultValue(),
+ overwrite,
+ Identifier.create("location_db", "location_table"),
+ staticPartitions,
+ null,
+ null,
+ partitionManager,
+ dynamicPartitionOverwrite);
+ }
+
private static Map<String, String> partitionSpec(String year, String month) {
LinkedHashMap<String, String> spec = new LinkedHashMap<>();
spec.put("year", year);
@@ -129,18 +343,69 @@ class FormatTableCommitRegistryValidationTest {
return spec;
}
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static void assertReplacementReport(
+ FormatTablePartitionManager partitionManager,
+ Path tablePath,
+ List<Map<String, String>> expectedSpecs) {
+ ArgumentCaptor<List<Map<String, String>>> specs =
+ ArgumentCaptor.forClass((Class) List.class);
+ ArgumentCaptor<List<PartitionStatistics>> statistics =
+ ArgumentCaptor.forClass((Class) List.class);
+ ArgumentCaptor<List<Map<String, String>>> options =
+ ArgumentCaptor.forClass((Class) List.class);
+ verify(partitionManager)
+ .createPartitions(
+ specs.capture(),
+ eq(true),
+ statistics.capture(),
+ eq(true),
+ options.capture());
+ assertThat(specs.getValue()).containsExactlyInAnyOrderElementsOf(expectedSpecs);
+ assertThat(statistics.getValue())
+ .extracting(PartitionStatistics::spec)
+ .containsExactlyInAnyOrderElementsOf(expectedSpecs);
+ List<Map<String, String>> reportedSpecs = specs.getValue();
+ assertThat(options.getValue()).hasSameSizeAs(expectedSpecs);
+ for (int i = 0; i < reportedSpecs.size(); i++) {
+ Map<String, String> spec = reportedSpecs.get(i);
+ assertThat(options.getValue().get(i))
+ .as("a replacement names the partition's own default directory")
+ .isEqualTo(
+ Collections.singletonMap(
+ PATH.key(),
+ new Path(
+ tablePath,
+ "year="
+ + spec.get("year")
+ + "/month="
+ + spec.get("month"))
+ .toString()));
+ }
+ }
+
private static class MutationTrackingLocalFileIO extends LocalFileIO {
private final AtomicInteger deleteCalls = new AtomicInteger();
private final AtomicInteger mkdirsCalls = new AtomicInteger();
+ private final Map<Path, AtomicInteger> listCalls = new LinkedHashMap<>();
private boolean tracking;
private void startTrackingMutations() {
deleteCalls.set(0);
mkdirsCalls.set(0);
+ listCalls.clear();
tracking = true;
}
+ @Override
+ public FileStatus[] listStatus(Path path) throws IOException {
+ if (tracking) {
+ listCalls.computeIfAbsent(path, ignored -> new AtomicInteger()).incrementAndGet();
+ }
+ return super.listStatus(path);
+ }
+
@Override
public boolean delete(Path path, boolean recursive) throws IOException {
if (tracking) {
@@ -164,5 +429,10 @@ class FormatTableCommitRegistryValidationTest {
private int mkdirsCalls() {
return mkdirsCalls.get();
}
+
+ private int listCalls(Path path) {
+ AtomicInteger calls = listCalls.get(path);
+ return calls == null ? 0 : calls.get();
+ }
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
index 1485f6148f..ef891793e1 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitStatisticsTest.java
@@ -98,8 +98,9 @@ class FormatTableCommitStatisticsTest {
.commit(Collections.singletonList(message));
long after = System.currentTimeMillis();
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.replaceStatistics).isFalse();
+ assertThat(reported.partitionOptions).isNull();
assertThat(reported.specs).containsExactly(spec("2025", "10"));
assertThat(reported.statistics).hasSize(1);
PartitionStatistics statistics = reported.statistics.get(0);
@@ -125,7 +126,7 @@ class FormatTableCommitStatisticsTest {
commit(tablePath, fileIO, partitionManager, false, null).commit(messages);
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.statistics)
.hasSize(2)
.anySatisfy(
@@ -159,7 +160,8 @@ class FormatTableCommitStatisticsTest {
commit(tablePath, fileIO, partitionManager, false, null).commit(messages);
- PartitionStatistics statistics = capture(partitionManager).statistics.get(0);
+ PartitionStatistics statistics =
+ captureAndValidateReport(partitionManager, tablePath).statistics.get(0);
// A sum missing a file must not be presented as an exact count.
assertThat(statistics.recordCount()).isEqualTo(PartitionStatistics.UNKNOWN);
assertThat(statistics.fileSizeInBytes()).isEqualTo(PartitionStatistics.UNKNOWN);
@@ -173,14 +175,20 @@ class FormatTableCommitStatisticsTest {
Path tablePath = new Path(tempDir.toUri());
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
// Something was there before this commit replaced it.
+ Path oldWrittenPartitionData = new Path(tablePath, "year=2025/month=10/old-data.csv");
writeDataFile(fileIO, tablePath, "year=2025/month=10", "old-data.csv", 4096);
CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128);
commit(tablePath, fileIO, partitionManager, true, null)
.commit(Collections.singletonList(message));
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.replaceStatistics).isTrue();
+ assertThat(reported.partitionOptions)
+ .containsExactly(returnsToDefault(tablePath, spec("2025", "10")));
+ assertThat(fileIO.exists(oldWrittenPartitionData)).isFalse();
+ verify(partitionManager, never()).listPartitions(any(), isNull());
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
assertThat(reported.statistics).hasSize(1);
PartitionStatistics statistics = reported.statistics.get(0);
assertThat(statistics.recordCount()).isEqualTo(3);
@@ -189,21 +197,50 @@ class FormatTableCommitStatisticsTest {
}
@Test
- void testStaticPrefixOverwriteZeroesAClearedPartitionAndKeepsItRegistered() throws Exception {
+ void testEmptyStaticOverwriteReportsZeroAndResetsLocationWithoutRegistryRead()
+ throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ Map<String, String> target = spec("2025", "10");
+
+ commit(tablePath, fileIO, partitionManager, true, target).commit(Collections.emptyList());
+
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
+ assertThat(reported.specs).containsExactly(target);
+ assertThat(reported.statistics).hasSize(1);
+ assertThat(reported.statistics.get(0).spec()).isEqualTo(target);
+ assertThat(reported.statistics.get(0).recordCount()).isZero();
+ assertThat(reported.statistics.get(0).fileSizeInBytes()).isZero();
+ assertThat(reported.statistics.get(0).fileCount()).isZero();
+ assertThat(reported.replaceStatistics).isTrue();
+ assertThat(reported.partitionOptions)
+ .containsExactly(returnsToDefault(tablePath, spec("2025", "10")));
+ verify(partitionManager, never()).listPartitions(any(), isNull());
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ }
+
+ @Test
+ void testStaticPrefixOverwriteReportsRegistryAndWrittenTargets() throws Exception {
LocalFileIO fileIO = LocalFileIO.create();
Path tablePath = new Path(tempDir.toUri());
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
- // Two sibling partitions hold data; the overwrite writes only one of them.
+ Map<String, String> prefix = Collections.singletonMap("year", "2025");
+ // The catalog has one unwritten partition with data and one whose directory is missing.
+ // The written partition is not registered yet, so the target set has to include both the
+ // prefix lookup and the commit messages.
+ registered(partitionManager, spec("2025", "11"), spec("2025", "12"), spec("2024", "09"));
+ Path oldWrittenPartitionData = new Path(tablePath, "year=2025/month=10/old-data.csv");
writeDataFile(fileIO, tablePath, "year=2025/month=10", "old-data.csv", 4096);
writeDataFile(fileIO, tablePath, "year=2025/month=11", "old-data.csv", 2048);
CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128);
long before = System.currentTimeMillis();
- commit(tablePath, fileIO, partitionManager, true, Collections.singletonMap("year", "2025"))
+ commit(tablePath, fileIO, partitionManager, true, prefix)
.commit(Collections.singletonList(message));
long after = System.currentTimeMillis();
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
long commitTime =
reported.statistics.stream()
.filter(s -> s.spec().equals(spec("2025", "10")))
@@ -212,9 +249,13 @@ class FormatTableCommitStatisticsTest {
.lastFileCreationTime();
assertThat(commitTime).isBetween(before, after);
assertThat(reported.replaceStatistics).isTrue();
+ assertThat(fileIO.exists(oldWrittenPartitionData)).isFalse();
// Red line: emptying a partition zeroes its statistics, it never unregisters it.
assertThat(reported.specs)
- .containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11"));
+ .containsExactlyInAnyOrder(
+ spec("2025", "10"), spec("2025", "11"), spec("2025", "12"));
+ verify(partitionManager).listPartitions(prefix, null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
verify(partitionManager, never()).dropPartitions(anyList());
assertThat(reported.statistics)
.anySatisfy(
@@ -227,6 +268,14 @@ class FormatTableCommitStatisticsTest {
// unknown would leave the stored one describing files that are gone,
// since an unknown replaces nothing.
assertThat(statistics.lastFileCreationTime()).isEqualTo(commitTime);
+ })
+ .anySatisfy(
+ statistics -> {
+ assertThat(statistics.spec()).isEqualTo(spec("2025", "12"));
+ assertThat(statistics.recordCount()).isZero();
+ assertThat(statistics.fileSizeInBytes()).isZero();
+ assertThat(statistics.fileCount()).isZero();
+ assertThat(statistics.lastFileCreationTime()).isEqualTo(commitTime);
});
}
@@ -244,7 +293,7 @@ class FormatTableCommitStatisticsTest {
overwritingTheWholeTable(tablePath, fileIO, partitionManager)
.commit(Collections.singletonList(message));
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.replaceStatistics).isTrue();
assertThat(reported.specs)
.containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11"));
@@ -265,6 +314,8 @@ class FormatTableCommitStatisticsTest {
assertThat(statistics.fileSizeInBytes()).isZero();
assertThat(statistics.fileCount()).isZero();
});
+ verify(partitionManager).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
}
@Test
@@ -286,8 +337,10 @@ class FormatTableCommitStatisticsTest {
.isTrue();
// Nor does the overwrite register it: reporting a zero for it would make a partition the
// catalog never had, out of a directory that still holds rows.
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.specs).containsExactly(spec("2025", "10"));
+ verify(partitionManager).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
}
@Test
@@ -295,25 +348,28 @@ class FormatTableCommitStatisticsTest {
LocalFileIO fileIO = LocalFileIO.create();
Path tablePath = new Path(tempDir.toUri());
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
- registered(partitionManager, spec("2025", "10"), spec("2025", "11"));
+ registered(partitionManager, spec("2025", "10"), spec("2025", "11"), spec("2024", "12"));
writeDataFile(fileIO, tablePath, "year=2025/month=10", "data.csv", 4096);
writeDataFile(fileIO, tablePath, "year=2025/month=11", "data.csv", 2048);
+ writeDataFile(fileIO, tablePath, "year=2024/month=12", "data.csv", 1024);
long before = System.currentTimeMillis();
commit(tablePath, fileIO, partitionManager, false, null)
.truncatePartitions(Arrays.asList(spec("2025", "10"), spec("2025", "11")));
long after = System.currentTimeMillis();
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
// What a truncated partition holds is zero, not zero fewer rows than before.
assertThat(reported.replaceStatistics).isTrue();
assertThat(reported.specs)
.containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11"));
// Red line: emptying a partition zeroes its statistics, it never unregisters it.
verify(partitionManager, never()).dropPartitions(anyList());
- // One authoritative request for the complete registry, then local selection.
- verify(partitionManager).listPartitions(Collections.emptyMap(), null);
- verify(partitionManager, never()).listPartitionsByNames(anyList());
+ // Complete names are resolved directly; truncating them never loads the full registry.
+ verify(partitionManager)
+ .listPartitionsByNames(Arrays.asList(spec("2025", "10"), spec("2025", "11")));
+ verify(partitionManager, never()).listPartitions(any(), isNull());
+ assertThat(fileIO.exists(new Path(tablePath, "year=2024/month=12/data.csv"))).isTrue();
// Statistics route by the spec they carry, so every partition needs its own.
assertThat(reported.statistics)
.extracting(PartitionStatistics::spec)
@@ -344,15 +400,17 @@ class FormatTableCommitStatisticsTest {
commit(tablePath, fileIO, partitionManager, false, null)
.truncatePartitions(Collections.singletonList(spec("2025", "10")));
- Reported reported = capture(partitionManager);
- // Unlike an overwrite, which reports only the files it removed itself, truncation states
- // that the partition holds nothing.
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
+ // A registered exact target remains a target even when it already holds no files.
assertThat(reported.replaceStatistics).isTrue();
assertThat(reported.specs).containsExactly(spec("2025", "10"));
assertThat(reported.statistics).hasSize(1);
assertThat(reported.statistics.get(0).spec()).isEqualTo(spec("2025", "10"));
assertThat(reported.statistics.get(0).recordCount()).isZero();
assertThat(reported.statistics.get(0).fileCount()).isZero();
+ verify(partitionManager)
+ .listPartitionsByNames(Collections.singletonList(spec("2025", "10")));
+ verify(partitionManager, never()).listPartitions(any(), isNull());
}
@Test
@@ -368,7 +426,7 @@ class FormatTableCommitStatisticsTest {
commit(tablePath, fileIO, partitionManager, false, null).truncateTable();
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.replaceStatistics).isTrue();
assertThat(reported.specs)
.containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11"));
@@ -378,6 +436,8 @@ class FormatTableCommitStatisticsTest {
assertThat(statistics.recordCount()).isZero();
assertThat(statistics.fileCount()).isZero();
});
+ verify(partitionManager).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
}
@Test
@@ -386,12 +446,10 @@ class FormatTableCommitStatisticsTest {
Path tablePath = new Path(tempDir.toUri());
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
Map<String, String> prefix = Collections.singletonMap("year", "2025");
- when(partitionManager.listPartitions(Collections.emptyMap(), null))
- .thenReturn(
- Arrays.asList(
- partition(spec("2025", "10")), partition(spec("2025", "11"))));
+ registered(partitionManager, spec("2025", "10"), spec("2025", "11"), spec("2024", "12"));
writeDataFile(fileIO, tablePath, "year=2025/month=10", "data.csv", 4096);
writeDataFile(fileIO, tablePath, "year=2025/month=11", "data.csv", 2048);
+ // Registered but outside the requested prefix.
writeDataFile(fileIO, tablePath, "year=2024/month=12", "data.csv", 1024);
// Under the prefix but not registered: a directory still waiting for MSCK REPAIR TABLE is
// not a partition of the table, so truncating neither empties nor registers it.
@@ -400,7 +458,7 @@ class FormatTableCommitStatisticsTest {
commit(tablePath, fileIO, partitionManager, false, null)
.truncatePartitions(Collections.singletonList(prefix));
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
// The prefix names no partition of its own; the partitions it empties are the registered
// ones underneath it.
assertThat(reported.specs)
@@ -409,6 +467,9 @@ class FormatTableCommitStatisticsTest {
assertThat(reported.statistics)
.allSatisfy(statistics -> assertThat(statistics.recordCount()).isZero());
assertThat(fileIO.exists(new Path(tablePath, "year=2025/month=12/data.csv"))).isTrue();
+ assertThat(fileIO.exists(new Path(tablePath, "year=2024/month=12/data.csv"))).isTrue();
+ verify(partitionManager).listPartitions(prefix, null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
}
@Test
@@ -424,13 +485,15 @@ class FormatTableCommitStatisticsTest {
commit(tablePath, fileIO, partitionManager, false, null).truncateTable();
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.replaceStatistics).isTrue();
assertThat(reported.specs)
.containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11"));
assertThat(reported.statistics)
.allSatisfy(statistics -> assertThat(statistics.fileCount()).isZero());
assertThat(fileIO.exists(new Path(tablePath, "year=2025/month=12/data.csv"))).isTrue();
+ verify(partitionManager).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
}
@Test
@@ -452,7 +515,7 @@ class FormatTableCommitStatisticsTest {
// A Format Table has no snapshot to make the whole truncation atomic, so what it emptied
// before the failure is reported anyway: the catalog must not keep describing files that
// are gone.
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.specs).containsExactly(spec("2025", "10"));
assertThat(reported.statistics.get(0).fileCount()).isZero();
}
@@ -510,11 +573,12 @@ class FormatTableCommitStatisticsTest {
}
@Test
- void testAClearedPartitionIsFoundEvenWhenTheListingAnswersUnderAnotherScheme()
+ void testARegisteredTargetIsReportedWhenFileListingAnswersUnderAnotherScheme()
throws Exception {
RescopingFileIO fileIO = new RescopingFileIO();
Path tablePath = new Path(tempDir.toUri());
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ registered(partitionManager, spec("2025", "11"));
writeDataFile(fileIO, tablePath, "year=2025/month=10", "old-data.csv", 4096);
writeDataFile(fileIO, tablePath, "year=2025/month=11", "old-data.csv", 2048);
CommitMessage message = writtenFile(fileIO, tablePath, "year=2025/month=10", 3, 128);
@@ -522,10 +586,10 @@ class FormatTableCommitStatisticsTest {
commit(tablePath, fileIO, partitionManager, true, Collections.singletonMap("year", "2025"))
.commit(Collections.singletonList(message));
- // A listing does not have to answer under the URI it was asked with, and matching whole
- // paths would then throw away a directory this very listing produced — leaving an emptied
- // partition holding stale statistics.
- assertThat(capture(partitionManager).statistics)
+ // The registry, not a filesystem path parsed after deletion, defines month=11 as a target.
+ // A listing may still answer under a different URI scheme, and deletion must handle that
+ // without losing the target's replacement statistics.
+ assertThat(captureAndValidateReport(partitionManager, tablePath).statistics)
.anySatisfy(
statistics -> {
assertThat(statistics.spec()).isEqualTo(spec("2025", "11"));
@@ -539,10 +603,10 @@ class FormatTableCommitStatisticsTest {
LocalFileIO fileIO = LocalFileIO.create();
Path tablePath = new Path(tempDir.toUri());
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
- // The key=value layout, where a directory that is not a partition of this table has no
- // spec at all rather than a plausible wrong one: the prefix directory itself, and a
- // directory nested below a partition. Clearing the prefix deletes the files in both,
- // because the listing collects data files at every level, not only the partition one.
+ registered(partitionManager, spec("2025", "11"));
+ // The registry contributes month=11 and the commit contributes month=10. A file directly
+ // under the prefix is outside both complete partition targets; a nested directory is data
+ // inside month=10, not a third partition inferred from its path.
writeDataFile(fileIO, tablePath, "year=2025/month=11", "old-data.csv", 2048);
writeDataFile(fileIO, tablePath, "year=2025", "orphan.csv", 512);
writeDataFile(fileIO, tablePath, "year=2025/month=10/nested", "old-data.csv", 1024);
@@ -551,13 +615,16 @@ class FormatTableCommitStatisticsTest {
commit(tablePath, fileIO, partitionManager, true, Collections.singletonMap("year", "2025"))
.commit(Collections.singletonList(message));
- // The commit succeeds and reports only the two real partitions. A directory with no spec
- // is left alone: its statistics go stale, which beats failing the commit that just wrote
- // the data, or accounting the files to a partition that does not exist.
- Reported reported = capture(partitionManager);
+ // The commit reports only the registry-defined and written targets. It leaves the foreign
+ // prefix file alone and deletes the nested file as part of month=10 without inventing a
+ // partition for either path.
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.specs)
.containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11"));
assertThat(reported.statistics).hasSize(2);
+ assertThat(fileIO.exists(new Path(tablePath, "year=2025/orphan.csv"))).isTrue();
+ assertThat(fileIO.exists(new Path(tablePath, "year=2025/month=10/nested/old-data.csv")))
+ .isFalse();
}
@Test
@@ -565,8 +632,10 @@ class FormatTableCommitStatisticsTest {
LocalFileIO fileIO = LocalFileIO.create();
Path tablePath = new Path(tempDir.toUri());
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
- // In the value-only layout a partition directory is the bare value, so the trailing two
- // components of 2025/10/nested read as the plausible partition {year=10, month=nested}.
+ registered(partitionManager, spec("2025", "11"));
+ // In the value-only layout the trailing components of 2025/10/nested could look like the
+ // plausible partition {year=10, month=nested}. Targets must still come only from the
+ // registry and the commit's written specs.
writeDataFile(fileIO, tablePath, "2025/10", "old-data.csv", 4096);
writeDataFile(fileIO, tablePath, "2025/10/nested", "old-data.csv", 4096);
writeDataFile(fileIO, tablePath, "2025/11", "old-data.csv", 2048);
@@ -581,9 +650,8 @@ class FormatTableCommitStatisticsTest {
true)
.commit(Collections.singletonList(message));
- // Only directories the spec rebuilds are reported: accounting 2025/10/nested to a partition
- // named {year=10, month=nested} would zero a partition this commit never touched.
- Reported reported = capture(partitionManager);
+ // The nested path is deleted as data under month=10, but is never reported as a partition.
+ Reported reported = captureAndValidateReport(partitionManager, tablePath, true);
assertThat(reported.specs)
.containsExactlyInAnyOrder(spec("2025", "10"), spec("2025", "11"));
assertThat(reported.statistics)
@@ -740,10 +808,19 @@ class FormatTableCommitStatisticsTest {
List<Partition> partitions = new ArrayList<>();
for (Map<String, String> spec : specs) {
partitions.add(partition(spec));
- when(partitionManager.listPartitions(spec, null))
- .thenReturn(Collections.singletonList(partition(spec)));
}
- when(partitionManager.listPartitions(Collections.emptyMap(), null)).thenReturn(partitions);
+ when(partitionManager.listPartitions(any(), isNull()))
+ .thenAnswer(
+ invocation -> {
+ Map<String, String> prefix = invocation.getArgument(0);
+ List<Partition> found = new ArrayList<>();
+ for (Partition partition : partitions) {
+ if (partition.spec().entrySet().containsAll(prefix.entrySet())) {
+ found.add(partition);
+ }
+ }
+ return found;
+ });
when(partitionManager.listPartitionsByNames(anyList()))
.thenAnswer(
invocation -> {
@@ -765,30 +842,61 @@ class FormatTableCommitStatisticsTest {
return spec;
}
+ /** What a replacement sends for a partition: the directory the partition belongs in. */
+ private static Map<String, String> returnsToDefault(Path tablePath, Map<String, String> spec) {
+ return returnsToDefault(tablePath, spec, false);
+ }
+
+ private static Map<String, String> returnsToDefault(
+ Path tablePath, Map<String, String> spec, boolean onlyValueInPath) {
+ return FormatTableCommitTest.defaultDirectoryOption(tablePath, spec, onlyValueInPath);
+ }
+
@SuppressWarnings({"unchecked", "rawtypes"})
- private static Reported capture(FormatTablePartitionManager partitionManager) {
+ private static Reported captureAndValidateReport(
+ FormatTablePartitionManager partitionManager, Path tablePath) {
+ return captureAndValidateReport(partitionManager, tablePath, false);
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static Reported captureAndValidateReport(
+ FormatTablePartitionManager partitionManager, Path tablePath, boolean onlyValueInPath) {
ArgumentCaptor<List<Map<String, String>>> specs =
ArgumentCaptor.forClass((Class) List.class);
ArgumentCaptor<List<PartitionStatistics>> statistics =
ArgumentCaptor.forClass((Class) List.class);
ArgumentCaptor<Boolean> replaceStatistics = ArgumentCaptor.forClass(Boolean.class);
+ ArgumentCaptor<List<Map<String, String>>> partitionOptions =
+ ArgumentCaptor.forClass((Class) List.class);
verify(partitionManager)
.createPartitions(
specs.capture(),
eq(true),
statistics.capture(),
replaceStatistics.capture(),
- isNull());
+ partitionOptions.capture());
+ List<Map<String, String>> capturedSpecs = specs.getValue();
+ List<Map<String, String>> capturedOptions = partitionOptions.getValue();
+ if (replaceStatistics.getValue()) {
+ assertThat(capturedOptions)
+ .as("replacement options must align with every target spec")
+ .hasSameSizeAs(capturedSpecs);
+ for (int i = 0; i < capturedSpecs.size(); i++) {
+ assertThat(capturedOptions.get(i))
+ .as("a replacement names the partition's own default directory")
+ .isEqualTo(
+ returnsToDefault(tablePath, capturedSpecs.get(i), onlyValueInPath));
+ }
+ } else {
+ assertThat(capturedOptions).as("append must not change partition options").isNull();
+ }
return new Reported(
- new ArrayList<>(specs.getValue()),
+ new ArrayList<>(capturedSpecs),
new ArrayList<>(statistics.getValue()),
- replaceStatistics.getValue());
+ replaceStatistics.getValue(),
+ capturedOptions == null ? null : new ArrayList<>(capturedOptions));
}
- /**
- * A {@link FileIO} that answers a listing with paths stripped of their scheme, the way a
- * delegating one does when it resolves the caller's scheme to the one it really uses.
- */
/** A file IO whose deletions all report failure, leaving the files in place. */
private static class RefusingFileIO extends LocalFileIO {
@@ -820,6 +928,10 @@ class FormatTableCommitStatisticsTest {
}
}
+ /**
+ * A {@link FileIO} that answers a listing with paths stripped of their scheme, the way a
+ * delegating one does when it resolves the caller's scheme to the one it really uses.
+ */
private static class RescopingFileIO extends LocalFileIO {
private static final long serialVersionUID = 1L;
@@ -978,7 +1090,7 @@ class FormatTableCommitStatisticsTest {
List<CommitMessage> messages = write.prepareCommit();
writeBuilder.newCommit().commit(messages);
- Reported reported = capture(partitionManager);
+ Reported reported = captureAndValidateReport(partitionManager, tablePath);
assertThat(reported.replaceStatistics).isFalse();
assertThat(reported.specs).containsExactly(Collections.singletonMap("year", "2025"));
assertThat(reported.statistics).hasSize(1);
@@ -1203,14 +1315,17 @@ class FormatTableCommitStatisticsTest {
private final List<Map<String, String>> specs;
private final List<PartitionStatistics> statistics;
private final boolean replaceStatistics;
+ @Nullable private final List<Map<String, String>> partitionOptions;
private Reported(
List<Map<String, String>> specs,
List<PartitionStatistics> statistics,
- boolean replaceStatistics) {
+ boolean replaceStatistics,
+ @Nullable List<Map<String, String>> partitionOptions) {
this.specs = specs;
this.statistics = statistics;
this.replaceStatistics = replaceStatistics;
+ this.partitionOptions = partitionOptions;
}
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
index 618ba4813c..24d957f97a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
@@ -195,21 +195,23 @@ class FormatTableCommitTest {
}
@Test
- void testStaticOverwriteRejectsCustomLocationBeforeDeletingOldData() throws Exception {
+ void testStaticOverwriteResetsCustomLocationWithoutDeletingExternalData() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "overwrite-custom-location");
Map<String, String> spec = Collections.singletonMap("part", "external");
Path oldData = new Path(tablePath, "part=external/data-old.csv");
+ Path customLocation =
+ new Path(new Path(tempDir.toUri()), "overwrite-custom-storage/part=external");
+ Path customData = new Path(customLocation, "data-external.csv");
fileIO.writeFile(oldData, "old", false);
+ fileIO.writeFile(customData, "external", false);
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
when(partitionManager.listPartitionsByNames(Collections.singletonList(spec)))
.thenReturn(
- Collections.singletonList(
- partitionAt(spec, "file:/external/part=external")));
+ Collections.singletonList(partitionAt(spec, customLocation.toString())));
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
- Collections.singletonList(
- partitionAt(spec, "file:/external/part=external")));
+ Collections.singletonList(partitionAt(spec, customLocation.toString())));
FormatTableCommit commit =
new FormatTableCommit(
tablePath.toString(),
@@ -226,29 +228,28 @@ class FormatTableCommitTest {
/* dynamicPartitionOverwrite */ true);
fileIO.startTrackingMutations();
- assertThatThrownBy(() -> commit.commit(Collections.emptyList()))
- .isInstanceOf(RuntimeException.class)
- .hasRootCauseInstanceOf(UnsupportedOperationException.class)
- .hasRootCauseMessage(
- "Overwriting catalog-managed Format Table partition {part=external} with "
- + "custom location 'file:/external/part=external' is not "
- + "supported.");
+ commit.commit(Collections.emptyList());
- assertThat(fileIO.exists(oldData)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.exists(oldData)).isFalse();
+ assertThat(fileIO.exists(customData)).isTrue();
+ assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ assertReplacementReport(partitionManager, tablePath, Collections.singletonList(spec));
}
@Test
- void testStaticOverwriteRejectsCustomLocationInsideTableRoot() throws Exception {
+ void testStaticOverwriteDoesNotReadUnrelatedRegistryRows() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "overwrite-future-default-location");
Map<String, String> customSpec = Collections.singletonMap("part", "external");
Map<String, String> targetSpec = Collections.singletonMap("part", "future");
Path targetPartitionPath = new Path(tablePath, "part=future");
- Path customData = new Path(targetPartitionPath, "data-custom.csv");
+ Path oldData = new Path(targetPartitionPath, "data-old.csv");
+ Path customLocation = new Path(tablePath, "unrelated-custom-location");
+ Path customData = new Path(customLocation, "data-custom.csv");
+ fileIO.writeFile(oldData, "old", false);
fileIO.writeFile(customData, "custom", false);
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
when(partitionManager.listPartitionsByNames(Collections.singletonList(targetSpec)))
@@ -256,7 +257,7 @@ class FormatTableCommitTest {
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
Collections.singletonList(
- partitionAt(customSpec, targetPartitionPath.toString())));
+ partitionAt(customSpec, customLocation.toString())));
FormatTableCommit commit =
new FormatTableCommit(
tablePath.toString(),
@@ -273,23 +274,19 @@ class FormatTableCommitTest {
/* dynamicPartitionOverwrite */ true);
fileIO.startTrackingMutations();
- Throwable failure = catchThrowable(() -> commit.commit(Collections.emptyList()));
- assertThat(failure).isInstanceOf(RuntimeException.class);
- assertThat(failure.getCause())
- .isInstanceOf(IllegalStateException.class)
- .hasMessage(
- "Catalog returned an invalid custom location for partition "
- + "{part=external} of Format Table location_db.location_table.");
+ commit.commit(Collections.emptyList());
+ assertThat(fileIO.exists(oldData)).isFalse();
assertThat(fileIO.exists(customData)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ assertReplacementReport(partitionManager, tablePath, Collections.singletonList(targetSpec));
}
@Test
- void testAppendRejectsCustomLocationInsideTableRoot() throws Exception {
+ void testAppendReadsOnlyWrittenPartitionNames() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "append-future-default-location");
Map<String, String> customSpec = Collections.singletonMap("part", "external");
@@ -323,31 +320,23 @@ class FormatTableCommitTest {
/* dynamicPartitionOverwrite */ true);
fileIO.startTrackingMutations();
- Throwable failure =
- catchThrowable(
- () ->
- commit.commit(
- Collections.singletonList(
- new TwoPhaseCommitMessage(committer))));
- assertThat(failure).isInstanceOf(RuntimeException.class);
- assertThat(failure.getCause())
- .isInstanceOf(IllegalStateException.class)
- .hasMessage(
- "Catalog returned an invalid custom location for partition "
- + "{part=external} of Format Table location_db.location_table.");
+ commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer)));
assertThat(fileIO.exists(customData)).isTrue();
assertThat(fileIO.exists(targetPath)).isFalse();
assertThat(fileIO.deleteCalls()).isZero();
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(committer, never()).commit(fileIO);
- verify(committer, never()).clean(fileIO);
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(committer).commit(fileIO);
+ verify(committer).clean(fileIO);
+ verify(partitionManager).listPartitionsByNames(Collections.singletonList(targetSpec));
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager).createPartitions(Collections.singletonList(targetSpec), true);
+ verify(partitionManager)
+ .createPartitions(anyList(), eq(true), anyList(), eq(false), isNull());
}
@Test
- void testAppendWithEmptyMessagesRejectsCustomLocationInsideStaticPrefix() throws Exception {
+ void testAppendWithEmptyMessagesDoesNotReadPartitionRegistry() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "append-partial-static-owned-prefix");
Map<String, String> staticPrefix = Collections.singletonMap("year", "2025");
@@ -374,19 +363,12 @@ class FormatTableCommitTest {
/* dynamicPartitionOverwrite */ true);
fileIO.startTrackingMutations();
- Throwable failure = catchThrowable(() -> commit.commit(Collections.emptyList()));
- assertThat(failure).isInstanceOf(RuntimeException.class);
- assertThat(failure.getCause())
- .isInstanceOf(IllegalStateException.class)
- .hasMessage(
- "Catalog returned an invalid custom location for partition "
- + "{year=2024, month=11} of Format Table "
- + "location_db.location_table.");
-
- assertThat(fileIO.exists(staticPrefixPath)).isFalse();
+ commit.commit(Collections.emptyList());
+
+ assertThat(fileIO.exists(staticPrefixPath)).isTrue();
assertThat(fileIO.deleteCalls()).isZero();
- assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager).listPartitions(Collections.emptyMap(), null);
+ assertThat(fileIO.mkdirsCalls()).isEqualTo(1);
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
verify(partitionManager, never()).listPartitions(staticPrefix, null);
verify(partitionManager, never()).listPartitionsByNames(anyList());
verify(partitionManager, never()).createPartitions(anyList(), anyBoolean());
@@ -395,23 +377,25 @@ class FormatTableCommitTest {
}
@Test
- void testDynamicOverwriteRejectsAffectedCustomLocationBeforeReplacingData() throws Exception {
+ void testDynamicOverwriteResetsCustomLocationWithoutDeletingExternalData() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "dynamic-overwrite-custom-location");
Map<String, String> spec = Collections.singletonMap("part", "external");
Path oldData = new Path(tablePath, "part=external/data-old.csv");
+ Path customLocation =
+ new Path(new Path(tempDir.toUri()), "dynamic-custom-storage/part=external");
+ Path customData = new Path(customLocation, "data-external.csv");
fileIO.writeFile(oldData, "old", false);
+ fileIO.writeFile(customData, "external", false);
TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class);
when(committer.targetPath()).thenReturn(new Path(tablePath, "part=external/data-new.csv"));
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
when(partitionManager.listPartitionsByNames(Collections.singletonList(spec)))
.thenReturn(
- Collections.singletonList(
- partitionAt(spec, "file:/external/part=external")));
+ Collections.singletonList(partitionAt(spec, customLocation.toString())));
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
- Collections.singletonList(
- partitionAt(spec, "file:/external/part=external")));
+ Collections.singletonList(partitionAt(spec, customLocation.toString())));
FormatTableCommit commit =
new FormatTableCommit(
tablePath.toString(),
@@ -428,37 +412,32 @@ class FormatTableCommitTest {
/* dynamicPartitionOverwrite */ true);
fileIO.startTrackingMutations();
- assertThatThrownBy(
- () ->
- commit.commit(
- Collections.singletonList(
- new TwoPhaseCommitMessage(committer))))
- .isInstanceOf(RuntimeException.class)
- .hasRootCauseInstanceOf(UnsupportedOperationException.class)
- .hasRootCauseMessage(
- "Overwriting catalog-managed Format Table partition {part=external} with "
- + "custom location 'file:/external/part=external' is not "
- + "supported.");
+ commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer)));
- assertThat(fileIO.exists(oldData)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.exists(oldData)).isFalse();
+ assertThat(fileIO.exists(customData)).isTrue();
+ assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(committer, never()).commit(fileIO);
- verify(committer, never()).clean(fileIO);
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(committer).commit(fileIO);
+ verify(committer).clean(fileIO);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ assertReplacementReport(partitionManager, tablePath, Collections.singletonList(spec));
}
@Test
- void testDynamicOverwriteRejectsCustomLocationInsideTableRoot() throws Exception {
+ void testDynamicOverwriteDoesNotReadUnrelatedRegistryRows() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath =
new Path(new Path(tempDir.toUri()), "dynamic-overwrite-future-default-location");
Map<String, String> customSpec = Collections.singletonMap("part", "external");
Map<String, String> targetSpec = Collections.singletonMap("part", "future");
Path targetPartitionPath = new Path(tablePath, "part=future");
- Path customData = new Path(targetPartitionPath, "data-custom.csv");
+ Path oldData = new Path(targetPartitionPath, "data-old.csv");
+ Path customLocation = new Path(tablePath, "unrelated-custom-location");
+ Path customData = new Path(customLocation, "data-custom.csv");
Path targetPath = new Path(targetPartitionPath, "data-new.csv");
+ fileIO.writeFile(oldData, "old", false);
fileIO.writeFile(customData, "custom", false);
TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class);
when(committer.targetPath()).thenReturn(targetPath);
@@ -468,7 +447,7 @@ class FormatTableCommitTest {
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
Collections.singletonList(
- partitionAt(customSpec, targetPartitionPath.toString())));
+ partitionAt(customSpec, customLocation.toString())));
FormatTableCommit commit =
new FormatTableCommit(
tablePath.toString(),
@@ -485,31 +464,22 @@ class FormatTableCommitTest {
/* dynamicPartitionOverwrite */ true);
fileIO.startTrackingMutations();
- Throwable failure =
- catchThrowable(
- () ->
- commit.commit(
- Collections.singletonList(
- new TwoPhaseCommitMessage(committer))));
- assertThat(failure).isInstanceOf(RuntimeException.class);
- assertThat(failure.getCause())
- .isInstanceOf(IllegalStateException.class)
- .hasMessage(
- "Catalog returned an invalid custom location for partition "
- + "{part=external} of Format Table location_db.location_table.");
+ commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer)));
+ assertThat(fileIO.exists(oldData)).isFalse();
assertThat(fileIO.exists(customData)).isTrue();
assertThat(fileIO.exists(targetPath)).isFalse();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(committer, never()).commit(fileIO);
- verify(committer, never()).clean(fileIO);
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(committer).commit(fileIO);
+ verify(committer).clean(fileIO);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ assertReplacementReport(partitionManager, tablePath, Collections.singletonList(targetSpec));
}
@Test
- void testStaticPrefixOverwriteRejectsCustomDescendantBeforeDeletingAnyPartition()
+ void testStaticPrefixOverwriteResetsEveryDescendantWithoutDeletingExternalData()
throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "prefix-overwrite-custom-location");
@@ -517,20 +487,24 @@ class FormatTableCommitTest {
Map<String, String> defaultSpec = partitionSpec("2025", "10");
Map<String, String> customSpec = partitionSpec("2025", "11");
Path defaultData = new Path(tablePath, "year=2025/month=10/data-old.csv");
- Path customResidue = new Path(tablePath, "year=2025/month=11/data-old.csv");
+ Path customDefaultData = new Path(tablePath, "year=2025/month=11/data-old.csv");
+ Path customLocation =
+ new Path(new Path(tempDir.toUri()), "prefix-custom-storage/year=2025/month=11");
+ Path externalData = new Path(customLocation, "data-external.csv");
fileIO.writeFile(defaultData, "default", false);
- fileIO.writeFile(customResidue, "residue", false);
+ fileIO.writeFile(customDefaultData, "old", false);
+ fileIO.writeFile(externalData, "external", false);
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
when(partitionManager.listPartitions(prefix, null))
.thenReturn(
Arrays.asList(
partitionAt(defaultSpec, null),
- partitionAt(customSpec, "file:/external/year=2025/month=11")));
+ partitionAt(customSpec, customLocation.toString())));
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
Arrays.asList(
partitionAt(defaultSpec, null),
- partitionAt(customSpec, "file:/external/year=2025/month=11")));
+ partitionAt(customSpec, customLocation.toString())));
FormatTableCommit commit =
new FormatTableCommit(
tablePath.toString(),
@@ -547,24 +521,22 @@ class FormatTableCommitTest {
/* dynamicPartitionOverwrite */ true);
fileIO.startTrackingMutations();
- assertThatThrownBy(() -> commit.commit(Collections.emptyList()))
- .isInstanceOf(RuntimeException.class)
- .hasRootCauseInstanceOf(UnsupportedOperationException.class)
- .hasRootCauseMessage(
- "Overwriting catalog-managed Format Table partition "
- + "{year=2025, month=11} with custom location "
- + "'file:/external/year=2025/month=11' is not supported.");
+ commit.commit(Collections.emptyList());
- assertThat(fileIO.exists(defaultData)).isTrue();
- assertThat(fileIO.exists(customResidue)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.exists(defaultData)).isFalse();
+ assertThat(fileIO.exists(customDefaultData)).isFalse();
+ assertThat(fileIO.exists(externalData)).isTrue();
+ assertThat(fileIO.deleteCalls()).isEqualTo(2);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(partitionManager).listPartitions(prefix, null);
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ assertReplacementReport(
+ partitionManager, tablePath, Arrays.asList(defaultSpec, customSpec));
}
@Test
- void testStaticPrefixOverwriteRejectsCustomLocationInsideTableRoot() throws Exception {
+ void testStaticPrefixOverwriteDoesNotReadUnrelatedRegistryRows() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath =
new Path(new Path(tempDir.toUri()), "prefix-overwrite-future-default-location");
@@ -596,20 +568,16 @@ class FormatTableCommitTest {
/* dynamicPartitionOverwrite */ true);
fileIO.startTrackingMutations();
- Throwable failure = catchThrowable(() -> commit.commit(Collections.emptyList()));
- assertThat(failure).isInstanceOf(RuntimeException.class);
- assertThat(failure.getCause())
- .isInstanceOf(IllegalStateException.class)
- .hasMessage(
- "Catalog returned an invalid custom location for partition "
- + "{year=2024, month=external} of Format Table "
- + "location_db.location_table.");
+ commit.commit(Collections.emptyList());
assertThat(fileIO.exists(customData)).isTrue();
assertThat(fileIO.deleteCalls()).isZero();
assertThat(fileIO.mkdirsCalls()).isZero();
+ verify(partitionManager).listPartitions(prefix, null);
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), anyList());
}
@Test
@@ -647,26 +615,31 @@ class FormatTableCommitTest {
assertThat(fileIO.exists(oldData)).isFalse();
assertThat(fileIO.exists(new Path(tablePath, "part=default"))).isTrue();
- verify(partitionManager)
- .createPartitions(anyList(), eq(true), anyList(), eq(true), isNull());
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ assertReplacementReport(partitionManager, tablePath, Collections.singletonList(targetSpec));
}
@Test
- void testWholeTableOverwriteRejectsCustomLocationBeforeDeletingAnyPartition() throws Exception {
+ void testWholeTableOverwriteResetsAllLocationsWithoutDeletingExternalData() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "whole-overwrite-custom-location");
Map<String, String> defaultSpec = Collections.singletonMap("part", "default");
Map<String, String> customSpec = Collections.singletonMap("part", "external");
Path defaultData = new Path(tablePath, "part=default/data-old.csv");
- Path customResidue = new Path(tablePath, "part=external/data-old.csv");
+ Path customDefaultData = new Path(tablePath, "part=external/data-old.csv");
+ Path customLocation =
+ new Path(new Path(tempDir.toUri()), "whole-custom-storage/part=external");
+ Path externalData = new Path(customLocation, "data-external.csv");
fileIO.writeFile(defaultData, "default", false);
- fileIO.writeFile(customResidue, "residue", false);
+ fileIO.writeFile(customDefaultData, "old", false);
+ fileIO.writeFile(externalData, "external", false);
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
Arrays.asList(
partitionAt(defaultSpec, null),
- partitionAt(customSpec, "file:/external/part=external")));
+ partitionAt(customSpec, customLocation.toString())));
FormatTableCommit commit =
new FormatTableCommit(
tablePath.toString(),
@@ -683,20 +656,17 @@ class FormatTableCommitTest {
/* dynamicPartitionOverwrite */ false);
fileIO.startTrackingMutations();
- assertThatThrownBy(() -> commit.commit(Collections.emptyList()))
- .isInstanceOf(RuntimeException.class)
- .hasRootCauseInstanceOf(UnsupportedOperationException.class)
- .hasRootCauseMessage(
- "Overwriting catalog-managed Format Table partition {part=external} with "
- + "custom location 'file:/external/part=external' is not "
- + "supported.");
+ commit.commit(Collections.emptyList());
- assertThat(fileIO.exists(defaultData)).isTrue();
- assertThat(fileIO.exists(customResidue)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.exists(defaultData)).isFalse();
+ assertThat(fileIO.exists(customDefaultData)).isFalse();
+ assertThat(fileIO.exists(externalData)).isTrue();
+ assertThat(fileIO.deleteCalls()).isEqualTo(2);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(partitionManager).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ assertReplacementReport(
+ partitionManager, tablePath, Arrays.asList(defaultSpec, customSpec));
}
@Test
@@ -1449,33 +1419,33 @@ class FormatTableCommitTest {
}
@Test
- void testTruncateTableRejectsCustomLocationBeforeDeletingOrReporting() throws Exception {
+ void testTruncateTableResetsCustomLocationWithoutDeletingExternalData() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "truncate-custom-location");
Map<String, String> spec = Collections.singletonMap("part", "external");
Path defaultData = new Path(tablePath, "part=external/data-old.csv");
+ Path customLocation =
+ new Path(new Path(tempDir.toUri()), "truncate-custom-storage/part=external");
+ Path externalData = new Path(customLocation, "data-external.csv");
fileIO.writeFile(defaultData, "old", false);
+ fileIO.writeFile(externalData, "external", false);
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
- Collections.singletonList(
- partitionAt(spec, "file:/external/part=external")));
+ Collections.singletonList(partitionAt(spec, customLocation.toString())));
FormatTableCommit commit =
truncatingCommit(tablePath, fileIO, false, partitionManager, "part");
fileIO.startTrackingMutations();
- assertThatThrownBy(commit::truncateTable)
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessage(
- "Truncating catalog-managed Format Table partition {part=external} with "
- + "custom location 'file:/external/part=external' is not "
- + "supported.");
+ commit.truncateTable();
- assertThat(fileIO.exists(defaultData)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.exists(defaultData)).isFalse();
+ assertThat(fileIO.exists(externalData)).isTrue();
+ assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(partitionManager).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ assertReplacementReport(partitionManager, tablePath, Collections.singletonList(spec));
}
@Test
@@ -1564,43 +1534,43 @@ class FormatTableCommitTest {
}
@Test
- void testTruncateNamedPartitionRejectsCustomLocationBeforeDeletingOrReporting()
+ void testTruncateNamedPartitionResetsCustomLocationWithoutDeletingExternalData()
throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "truncate-named-custom-location");
Map<String, String> spec = partitionSpec("2025", "10");
Path defaultData = new Path(tablePath, "year=2025/month=10/data-old.csv");
+ Path customLocation =
+ new Path(
+ new Path(tempDir.toUri()),
+ "truncate-named-custom-storage/year=2025/month=10");
+ Path externalData = new Path(customLocation, "data-external.csv");
fileIO.writeFile(defaultData, "old", false);
+ fileIO.writeFile(externalData, "external", false);
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
when(partitionManager.listPartitionsByNames(Collections.singletonList(spec)))
.thenReturn(
- Collections.singletonList(
- partitionAt(spec, "file:/external/year=2025/month=10")));
+ Collections.singletonList(partitionAt(spec, customLocation.toString())));
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
- Collections.singletonList(
- partitionAt(spec, "file:/external/year=2025/month=10")));
+ Collections.singletonList(partitionAt(spec, customLocation.toString())));
FormatTableCommit commit =
truncatingCommit(tablePath, fileIO, false, partitionManager, "year", "month");
fileIO.startTrackingMutations();
- assertThatThrownBy(() -> commit.truncatePartitions(Collections.singletonList(spec)))
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessage(
- "Truncating catalog-managed Format Table partition "
- + "{year=2025, month=10} with custom location "
- + "'file:/external/year=2025/month=10' is not supported.");
+ commit.truncatePartitions(Collections.singletonList(spec));
- assertThat(fileIO.exists(defaultData)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.exists(defaultData)).isFalse();
+ assertThat(fileIO.exists(externalData)).isTrue();
+ assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(partitionManager).listPartitionsByNames(Collections.singletonList(spec));
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ assertReplacementReport(partitionManager, tablePath, Collections.singletonList(spec));
}
@Test
- void testTruncateNamedPartitionRejectsIncompleteSpecFromFullRegistryBeforeMutation()
- throws Exception {
+ void testTruncateNamedPartitionDoesNotReadFullRegistry() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "truncate-named-incomplete-spec");
Map<String, String> targetSpec = partitionSpec("2025", "10");
@@ -1608,104 +1578,104 @@ class FormatTableCommitTest {
Path oldData = new Path(tablePath, "year=2025/month=10/data-old.csv");
fileIO.writeFile(oldData, "old", false);
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(incompleteSpec, null)));
FormatTableCommit commit =
truncatingCommit(tablePath, fileIO, false, partitionManager, "year", "month");
fileIO.startTrackingMutations();
- assertThatThrownBy(() -> commit.truncatePartitions(Collections.singletonList(targetSpec)))
- .isInstanceOf(IllegalStateException.class)
- .hasMessage(
- "Catalog returned incomplete partition spec {year=2025} for Format Table "
- + "truncate_db.truncate_table.");
+ commit.truncatePartitions(Collections.singletonList(targetSpec));
- assertThat(fileIO.exists(oldData)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.exists(oldData)).isFalse();
+ assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager).listPartitions(Collections.emptyMap(), null);
- verify(partitionManager, never()).listPartitionsByNames(anyList());
+ verify(partitionManager).listPartitionsByNames(Collections.singletonList(targetSpec));
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
verify(partitionManager, never()).createPartitions(anyList(), anyBoolean());
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ assertReplacementReport(partitionManager, tablePath, Collections.singletonList(targetSpec));
}
@Test
- void testTruncateNamedPartitionRejectsCustomLocationInsideTableRootBeforeMutation()
- throws Exception {
+ void testTruncateNamedPartitionLeavesUnrelatedCustomLocationUntouched() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "truncate-named-overlapping-location");
Map<String, String> targetSpec = partitionSpec("2025", "10");
Map<String, String> customSpec = partitionSpec("2024", "11");
Path targetPartitionPath = new Path(tablePath, "year=2025/month=10");
- Path customData = new Path(targetPartitionPath, "data-custom.csv");
+ Path targetData = new Path(targetPartitionPath, "data-old.csv");
+ Path customLocation = new Path(tablePath, "unrelated-custom-location");
+ Path customData = new Path(customLocation, "data-custom.csv");
+ fileIO.writeFile(targetData, "old", false);
fileIO.writeFile(customData, "custom", false);
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ when(partitionManager.listPartitionsByNames(Collections.singletonList(targetSpec)))
+ .thenReturn(Collections.singletonList(partitionAt(targetSpec, null)));
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
Arrays.asList(
partitionAt(targetSpec, null),
- partitionAt(customSpec, targetPartitionPath.toString())));
+ partitionAt(customSpec, customLocation.toString())));
FormatTableCommit commit =
truncatingCommit(tablePath, fileIO, false, partitionManager, "year", "month");
fileIO.startTrackingMutations();
- assertThatThrownBy(() -> commit.truncatePartitions(Collections.singletonList(targetSpec)))
- .isInstanceOf(IllegalStateException.class)
- .hasMessage(
- "Catalog returned an invalid custom location for partition "
- + "{year=2024, month=11} of Format Table "
- + "truncate_db.truncate_table.");
+ commit.truncatePartitions(Collections.singletonList(targetSpec));
+ assertThat(fileIO.exists(targetData)).isFalse();
assertThat(fileIO.exists(customData)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.deleteCalls()).isEqualTo(1);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager).listPartitions(Collections.emptyMap(), null);
- verify(partitionManager, never()).listPartitionsByNames(anyList());
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(partitionManager).listPartitionsByNames(Collections.singletonList(targetSpec));
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ assertReplacementReport(partitionManager, tablePath, Collections.singletonList(targetSpec));
}
@Test
- void testTruncatePrefixRejectsCustomDescendantBeforeMutatingDefaultDescendant()
- throws Exception {
+ void testTruncatePrefixResetsEveryDescendantWithoutDeletingExternalData() throws Exception {
MutationTrackingLocalFileIO fileIO = new MutationTrackingLocalFileIO();
Path tablePath = new Path(new Path(tempDir.toUri()), "truncate-prefix-custom-location");
Map<String, String> prefix = Collections.singletonMap("year", "2025");
Map<String, String> defaultSpec = partitionSpec("2025", "10");
Map<String, String> customSpec = partitionSpec("2025", "11");
Path defaultData = new Path(tablePath, "year=2025/month=10/data-old.csv");
- Path customResidue = new Path(tablePath, "year=2025/month=11/data-old.csv");
+ Path customDefaultData = new Path(tablePath, "year=2025/month=11/data-old.csv");
+ Path customLocation =
+ new Path(
+ new Path(tempDir.toUri()),
+ "truncate-prefix-custom-storage/year=2025/month=11");
+ Path externalData = new Path(customLocation, "data-external.csv");
fileIO.writeFile(defaultData, "default", false);
- fileIO.writeFile(customResidue, "residue", false);
+ fileIO.writeFile(customDefaultData, "old", false);
+ fileIO.writeFile(externalData, "external", false);
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
when(partitionManager.listPartitions(prefix, null))
.thenReturn(
Arrays.asList(
partitionAt(defaultSpec, null),
- partitionAt(customSpec, "file:/external/year=2025/month=11")));
+ partitionAt(customSpec, customLocation.toString())));
when(partitionManager.listPartitions(Collections.emptyMap(), null))
.thenReturn(
Arrays.asList(
partitionAt(defaultSpec, null),
- partitionAt(customSpec, "file:/external/year=2025/month=11")));
+ partitionAt(customSpec, customLocation.toString())));
FormatTableCommit commit =
truncatingCommit(tablePath, fileIO, false, partitionManager, "year", "month");
fileIO.startTrackingMutations();
- assertThatThrownBy(() -> commit.truncatePartitions(Collections.singletonList(prefix)))
- .isInstanceOf(UnsupportedOperationException.class)
- .hasMessage(
- "Truncating catalog-managed Format Table partition "
- + "{year=2025, month=11} with custom location "
- + "'file:/external/year=2025/month=11' is not supported.");
+ commit.truncatePartitions(Collections.singletonList(prefix));
- assertThat(fileIO.exists(defaultData)).isTrue();
- assertThat(fileIO.exists(customResidue)).isTrue();
- assertThat(fileIO.deleteCalls()).isZero();
+ assertThat(fileIO.exists(defaultData)).isFalse();
+ assertThat(fileIO.exists(customDefaultData)).isFalse();
+ assertThat(fileIO.exists(externalData)).isTrue();
+ assertThat(fileIO.deleteCalls()).isEqualTo(2);
assertThat(fileIO.mkdirsCalls()).isZero();
- verify(partitionManager, never())
- .createPartitions(anyList(), anyBoolean(), any(), anyBoolean(), isNull());
+ verify(partitionManager).listPartitions(prefix, null);
+ verify(partitionManager, never()).listPartitions(Collections.emptyMap(), null);
+ verify(partitionManager, never()).listPartitionsByNames(anyList());
+ assertReplacementReport(
+ partitionManager, tablePath, Arrays.asList(defaultSpec, customSpec));
}
@Test
@@ -2768,13 +2738,23 @@ class FormatTableCommitTest {
@SuppressWarnings({"unchecked", "rawtypes"})
@Test
- void testCleanupStatisticsClaimOnlyFilesDeletedByThisCommit() throws Exception {
+ void testOverwriteReportsEveryRegisteredTargetAfterCleanup() throws Exception {
MixedOwnershipFileIO fileIO = new MixedOwnershipFileIO();
Path tablePath = new Path(tempDir.toUri());
writeOldFiles(fileIO, new Path(tablePath, "year=2025/month=00"), 1);
writeOldFiles(fileIO, new Path(tablePath, "year=2025/month=01"), 1);
writeOldFiles(fileIO, new Path(tablePath, "year=2025/month=02"), 1);
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ List<Map<String, String>> expectedSpecs =
+ Arrays.asList(
+ partitionSpec("2025", "00"),
+ partitionSpec("2025", "01"),
+ partitionSpec("2025", "02"));
+ when(partitionManager.listPartitions(Collections.singletonMap("year", "2025"), null))
+ .thenReturn(
+ expectedSpecs.stream()
+ .map(spec -> partitionAt(spec, null))
+ .collect(java.util.stream.Collectors.toList()));
FormatTableCommit commit =
new FormatTableCommit(
tablePath.toString(),
@@ -2794,20 +2774,11 @@ class FormatTableCommitTest {
commit.commit(Collections.emptyList());
- Map<String, String> owned = partitionSpec("2025", "00");
- ArgumentCaptor<List<Map<String, String>>> specs =
- ArgumentCaptor.forClass((Class) List.class);
- ArgumentCaptor<List<PartitionStatistics>> statistics =
- ArgumentCaptor.forClass((Class) List.class);
- verify(partitionManager)
- .createPartitions(
- specs.capture(), eq(true), statistics.capture(), eq(true), isNull());
- assertThat(specs.getValue()).containsExactly(owned);
- assertThat(statistics.getValue())
- .singleElement()
- .satisfies(
+ List<PartitionStatistics> statistics =
+ assertReplacementReport(partitionManager, tablePath, expectedSpecs);
+ assertThat(statistics)
+ .allSatisfy(
stat -> {
- assertThat(stat.spec()).isEqualTo(owned);
assertThat(stat.recordCount()).isZero();
assertThat(stat.fileSizeInBytes()).isZero();
assertThat(stat.fileCount()).isZero();
@@ -2851,13 +2822,22 @@ class FormatTableCommitTest {
fileIO, new Path(tablePath, String.format("year=2025/month=%02d", month)), 1);
}
FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ List<Map<String, String>> expectedSpecs = new ArrayList<>();
+ for (int month = 0; month < 8; month++) {
+ expectedSpecs.add(partitionSpec("2025", String.format("%02d", month)));
+ }
+ when(partitionManager.listPartitions(Collections.singletonMap("year", "2025"), null))
+ .thenReturn(
+ expectedSpecs.stream()
+ .map(spec -> partitionAt(spec, null))
+ .collect(java.util.stream.Collectors.toList()));
doAnswer(
invocation -> {
assertThat(fileIO.activeDeletes()).isZero();
return null;
})
.when(partitionManager)
- .createPartitions(anyList(), eq(true), anyList(), eq(true), isNull());
+ .createPartitions(anyList(), eq(true), anyList(), eq(true), anyList());
FormatTableCommit commit =
new FormatTableCommit(
tablePath.toString(),
@@ -2877,23 +2857,13 @@ class FormatTableCommitTest {
commit.commit(Collections.emptyList());
- List<Map<String, String>> expectedSpecs = new ArrayList<>();
- for (int month = 0; month < 8; month++) {
- expectedSpecs.add(partitionSpec("2025", String.format("%02d", month)));
- }
- ArgumentCaptor<List<Map<String, String>>> specs =
- ArgumentCaptor.forClass((Class) List.class);
- ArgumentCaptor<List<PartitionStatistics>> statistics =
- ArgumentCaptor.forClass((Class) List.class);
- verify(partitionManager)
- .createPartitions(
- specs.capture(), eq(true), statistics.capture(), eq(true), isNull());
- assertThat(specs.getValue()).containsExactlyInAnyOrderElementsOf(expectedSpecs);
- assertThat(statistics.getValue())
+ List<PartitionStatistics> statistics =
+ assertReplacementReport(partitionManager, tablePath, expectedSpecs);
+ assertThat(statistics)
.hasSize(8)
.extracting(PartitionStatistics::spec)
.containsExactlyInAnyOrderElementsOf(expectedSpecs);
- assertThat(statistics.getValue())
+ assertThat(statistics)
.allSatisfy(
stat -> {
assertThat(stat.recordCount()).isZero();
@@ -3883,6 +3853,63 @@ class FormatTableCommitTest {
return spec;
}
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static List<PartitionStatistics> assertReplacementReport(
+ FormatTablePartitionManager partitionManager,
+ Path tablePath,
+ List<Map<String, String>> expectedSpecs) {
+ return assertReplacementReport(partitionManager, tablePath, false, expectedSpecs);
+ }
+
+ /**
+ * What a replacement sends for a partition: the directory that partition belongs in, named the
+ * way partition directories are named, escapes and all.
+ */
+ static Map<String, String> defaultDirectoryOption(
+ Path tablePath, Map<String, String> spec, boolean onlyValueInPath) {
+ return Collections.singletonMap(
+ CoreOptions.PATH.key(),
+ new Path(
+ tablePath,
+ PartitionPathUtils.generatePartitionPathUtil(
+ new LinkedHashMap<>(spec), onlyValueInPath))
+ .toString());
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static List<PartitionStatistics> assertReplacementReport(
+ FormatTablePartitionManager partitionManager,
+ Path tablePath,
+ boolean onlyValueInPath,
+ List<Map<String, String>> expectedSpecs) {
+ ArgumentCaptor<List<Map<String, String>>> specs =
+ ArgumentCaptor.forClass((Class) List.class);
+ ArgumentCaptor<List<PartitionStatistics>> statistics =
+ ArgumentCaptor.forClass((Class) List.class);
+ ArgumentCaptor<List<Map<String, String>>> options =
+ ArgumentCaptor.forClass((Class) List.class);
+ verify(partitionManager)
+ .createPartitions(
+ specs.capture(),
+ eq(true),
+ statistics.capture(),
+ eq(true),
+ options.capture());
+ assertThat(specs.getValue()).containsExactlyInAnyOrderElementsOf(expectedSpecs);
+ assertThat(statistics.getValue())
+ .extracting(PartitionStatistics::spec)
+ .containsExactlyInAnyOrderElementsOf(expectedSpecs);
+ assertThat(options.getValue()).hasSameSizeAs(specs.getValue());
+ for (int i = 0; i < specs.getValue().size(); i++) {
+ assertThat(options.getValue().get(i))
+ .as("a replacement names the partition's own default directory")
+ .isEqualTo(
+ defaultDirectoryOption(
+ tablePath, specs.getValue().get(i), onlyValueInPath));
+ }
+ return statistics.getValue();
+ }
+
/** The commit TRUNCATE makes: nothing to write, so no overwrite and no static partition. */
private FormatTableCommit truncatingCommit(
Path tableLocation,
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionPathResolverTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionPathResolverTest.java
index 3bfbe3188a..f85f043d4c 100644
--- a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionPathResolverTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionPathResolverTest.java
@@ -47,6 +47,57 @@ class FormatTablePartitionPathResolverTest {
private static final Path TABLE_PATH = new Path("file:/warehouse/table");
private static final String TABLE_NAME = "db.table";
+ @Test
+ void testADefaultDirectoryIsRecognizedWhateverItsSchemeOrEscapes() {
+ // The table's own directory is where a partition goes back to, so recognizing it may not
+ // depend on rules written for a location someone typed: a scheme without an authority is
+ // a supported table location, and an escaped value is one directory name.
+ LinkedHashMap<String, String> escaped = new LinkedHashMap<>();
+ escaped.put("dt", "a%b");
+ escaped.put("hh", "a/b");
+ for (Path tablePath :
+ Arrays.asList(
+ new Path("viewfs:/warehouse/table"),
+ new Path("cfs:/data/warehouse/table"),
+ new Path("oss://bucket/warehouse/table"))) {
+ Path defaultPath =
+ FormatTablePartitionPathResolver.defaultPartitionPath(
+ tablePath, escaped, false);
+ assertThat(defaultPath.toString()).contains("dt=a%25b/hh=a%2Fb");
+ assertThat(
+ FormatTablePartitionPathResolver.isDefaultPartitionPath(
+ tablePath, escaped, false, defaultPath.toString(), null))
+ .as("%s must recognize its own partition directory", tablePath)
+ .isTrue();
+ assertThat(
+ FormatTablePartitionPathResolver.isDefaultPartitionPath(
+ tablePath,
+ escaped,
+ false,
+ new Path(tablePath, "dt=a%25b").toString(),
+ null))
+ .as("%s must not mistake another directory for it", tablePath)
+ .isFalse();
+ }
+ }
+
+ @Test
+ void testATableDirectoryNeedsNoAuthorityButAPartitionOwnsNoSuchPlace() {
+ // A catalog whose warehouse has no authority still names one directory per partition, and
+ // a request returns a partition there by naming it; what a partition may own is stricter.
+ assertThat(
+ FormatTablePartitionPathResolver.canonicalizeLocation(
+ "cfs:/data/warehouse/db/table/dt=2026", null)
+ .toString())
+ .isEqualTo("cfs:/data/warehouse/db/table/dt=2026");
+ assertThatThrownBy(
+ () ->
+ FormatTablePartitionPathResolver.canonicalizeCustomLocation(
+ "cfs:/data/warehouse/db/table/dt=2026", null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid custom partition location");
+ }
+
@ParameterizedTest
@ValueSource(
strings = {