> p
throws TableNotExistException {}
/**
- * Create partitions of the specify table, with explicit existence semantics and optionally
- * reporting statistics for them in the same call.
- *
- * The statistics are matched to {@code partitions} by {@link PartitionStatistics#spec()}, so
- * they may cover only some of them, and {@code replaceStatistics} says whether they replace
- * what the catalog already holds or add to it. What decides whether they survive is whether a
- * catalog overrides this method: one that does not registers the partitions exactly as {@link
- * #createPartitions(Identifier, List)} does and drops the report, however much of it the
- * catalog could have stored, and for a catalog that keeps no partitions at all that means it
- * does nothing.
- *
- * @param identifier path of the table to create partitions
- * @param partitions partitions to be created
- * @param ignoreIfExists if false, fail when any partition already exists and apply none of the
- * batch; if true, behave like {@link #createPartitions(Identifier, List)}
- * @param statistics statistics to report, or null to report none
- * @param replaceStatistics whether the report replaces the stored values rather than adding to
- * them; ignored when {@code statistics} is null
- * @throws TableNotExistException if the table does not exist
- * @throws UnsupportedOperationException if {@code ignoreIfExists} is false and the catalog does
- * not implement strict creation, which is what the default here does
+ * Create partitions atomically unless existing entries are ignored, with optional statistics
+ * and position-aligned locations whose null entries use defaults.
*/
default void createPartitions(
Identifier identifier,
List> partitions,
boolean ignoreIfExists,
@Nullable List statistics,
- boolean replaceStatistics)
+ boolean replaceStatistics,
+ @Nullable List partitionLocations)
throws TableNotExistException {
+ if (partitionLocations != null) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Catalog %s does not support partition locations.",
+ getClass().getName()));
+ }
if (!ignoreIfExists) {
throw new UnsupportedOperationException(
String.format(
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
index 6c691e6cee28..fcbe9f6608d9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -331,10 +331,16 @@ public void createPartitions(
List> partitions,
boolean ignoreIfExists,
@Nullable List statistics,
- boolean replaceStatistics)
+ boolean replaceStatistics,
+ @Nullable List partitionLocations)
throws TableNotExistException {
wrapped.createPartitions(
- identifier, partitions, ignoreIfExists, statistics, replaceStatistics);
+ identifier,
+ partitions,
+ ignoreIfExists,
+ statistics,
+ replaceStatistics,
+ partitionLocations);
}
@Override
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 50fe373d6a03..342e7bf408a4 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
@@ -67,6 +67,7 @@
import org.apache.paimon.table.Instant;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableSnapshot;
+import org.apache.paimon.table.format.FormatTablePartitionPathResolver;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.system.SystemTableLoader;
import org.apache.paimon.utils.JsonSerdeUtil;
@@ -760,7 +761,7 @@ public void markDonePartitions(Identifier identifier, List>
@Override
public void createPartitions(Identifier identifier, List> partitions)
throws TableNotExistException {
- createPartitions(identifier, partitions, true, null, false);
+ createPartitions(identifier, partitions, true, null, false, null);
}
@Override
@@ -769,11 +770,19 @@ public void createPartitions(
List> partitions,
boolean ignoreIfExists,
@Nullable List statistics,
- boolean replaceStatistics)
+ boolean replaceStatistics,
+ @Nullable List partitionLocations)
throws TableNotExistException {
+ List canonicalLocations =
+ canonicalizePartitionLocations(identifier, partitions, partitionLocations);
try {
api.createPartitions(
- identifier, partitions, ignoreIfExists, statistics, replaceStatistics);
+ identifier,
+ partitions,
+ ignoreIfExists,
+ statistics,
+ replaceStatistics,
+ canonicalLocations);
} catch (NoSuchResourceException e) {
throw new TableNotExistException(identifier);
} catch (ForbiddenException e) {
@@ -786,7 +795,60 @@ public void createPartitions(
identifier, e.getMessage()));
} catch (BadRequestException e) {
throw new IllegalArgumentException(e.getMessage());
+ } catch (NotImplementedException e) {
+ if (canonicalLocations == null) {
+ throw e;
+ }
+ throw new UnsupportedOperationException(
+ String.format(
+ "REST Catalog server does not support custom partition locations for table %s.",
+ identifier.getFullName()),
+ e);
+ }
+ }
+
+ @Nullable
+ private List canonicalizePartitionLocations(
+ Identifier identifier,
+ List> partitions,
+ @Nullable List requested) {
+ if (requested == null) {
+ return null;
+ }
+ if (requested.size() != partitions.size()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition locations for table %s must align with all %d partition specs, but found %d.",
+ identifier.getFullName(), partitions.size(), requested.size()));
+ }
+ List canonical = new ArrayList<>(requested.size());
+ for (int i = 0; i < requested.size(); i++) {
+ String location = requested.get(i);
+ if (location == null) {
+ canonical.add(null);
+ continue;
+ }
+ try {
+ canonical.add(
+ FormatTablePartitionPathResolver.canonicalizeCustomLocation(
+ location, context)
+ .toString());
+ } catch (IllegalArgumentException e) {
+ throw invalidPartitionLocation(identifier, partitions.get(i), e);
+ }
}
+ return canonical;
+ }
+
+ private static IllegalArgumentException invalidPartitionLocation(
+ Identifier identifier,
+ @Nullable Map partition,
+ IllegalArgumentException cause) {
+ String message =
+ String.format(
+ "Invalid custom partition location for partition %s of table %s.",
+ partition, identifier.getFullName());
+ return new IllegalArgumentException(message, cause);
}
@Override
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 1f4d4575a4ff..fc7f70dc4c54 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
@@ -159,10 +159,12 @@ public void createPartitions(
List> partitions,
boolean ignoreIfExists,
@Nullable List statistics,
- boolean replaceStatistics) {
+ boolean replaceStatistics,
+ @Nullable List partitionLocations) {
// Validated before the empty check: returning early would swallow a malformed report.
Map, PartitionStatistics> statisticsBySpec =
validateAndIndexStatistics(statistics, partitions);
+ validateLocations(partitionLocations, partitions);
if (partitions.isEmpty()) {
return;
}
@@ -172,19 +174,31 @@ public void createPartitions(
// Rejecting the whole batch when any partition exists is only meaningful
// if the batch stays one request, so a strict create is never split.
catalog.createPartitions(
- identifier, partitions, false, statistics, replaceStatistics);
+ identifier,
+ partitions,
+ false,
+ statistics,
+ replaceStatistics,
+ partitionLocations);
return null;
}
// isRetrySafe() bounds the transport retry only: a caller-level rerun of a
// multi-batch ADD still double counts the batches that already landed.
- for (List> batch : batches(partitions)) {
+ for (int start = 0; start < partitions.size(); start += REQUEST_SIZE) {
+ int end = Math.min(start + REQUEST_SIZE, partitions.size());
+ List> batch = partitions.subList(start, end);
+ List batchLocations =
+ partitionLocations == null
+ ? null
+ : partitionLocations.subList(start, end);
// A partition and its statistics travel in the same request.
catalog.createPartitions(
identifier,
batch,
true,
statisticsOf(batch, statisticsBySpec),
- replaceStatistics);
+ replaceStatistics,
+ batchLocations);
}
return null;
},
@@ -237,6 +251,37 @@ private Map, PartitionStatistics> validateAndIndexStatistics
return bySpec;
}
+ private void validateLocations(
+ @Nullable List locations, List> partitions) {
+ if (locations == null) {
+ return;
+ }
+ String tableName = identifier.getFullName();
+ checkArgument(
+ locations.size() == partitions.size(),
+ "Partition locations for table %s must contain the same number of entries as "
+ + "partitions: %s locations for %s partitions.",
+ tableName,
+ locations.size(),
+ partitions.size());
+ Set> registered = capacityFor(partitions.size());
+ for (int i = 0; i < partitions.size(); i++) {
+ Map spec = partitions.get(i);
+ checkArgument(
+ registered.add(spec),
+ "Partition %s of table %s is registered twice in one request with an "
+ + "aligned location list; register each partition once.",
+ spec,
+ tableName);
+ String location = locations.get(i);
+ checkArgument(
+ location == null || !StringUtils.isBlank(location),
+ "Location for partition %s of table %s is blank.",
+ spec,
+ tableName);
+ }
+ }
+
@Nullable
private static List statisticsOf(
List> batch,
@@ -246,9 +291,9 @@ private static List statisticsOf(
}
List ofBatch = new ArrayList<>(batch.size());
for (Map spec : batch) {
- PartitionStatistics statistic = statisticsBySpec.get(spec);
- if (statistic != null) {
- ofBatch.add(statistic);
+ PartitionStatistics statistics = statisticsBySpec.get(spec);
+ if (statistics != null) {
+ ofBatch.add(statistics);
}
}
return ofBatch;
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
index 9095df296fee..35083294220c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogSplitEnumerator.java
@@ -20,7 +20,6 @@
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.PartitionEntry;
@@ -89,7 +88,8 @@ final class CatalogSplitEnumerator extends SplitEnumerator {
@Override
List enumeratePartitions(@Nullable PartitionPredicate partitionFilter)
throws IOException {
- return enumeratePartitions(findCatalogPartitions(partitionFilter), partitionFilter);
+ CatalogPartitionListing listing = findCatalogPartitions(partitionFilter);
+ return enumeratePartitions(filterPartitions(listing.partitionPaths, partitionFilter));
}
@Override
@@ -97,34 +97,43 @@ ScanPlan plan(@Nullable PartitionPredicate partitionFilter) throws IOException {
if (table.partitionKeys().isEmpty()) {
return super.plan(partitionFilter);
}
- List partitions = findCatalogPartitions(partitionFilter);
- List entries = toPartitionEntries(partitions, partitionFilter);
- return new ScanPlan(enumeratePartitions(partitions, partitionFilter), rowCount(entries));
+ CatalogPartitionListing listing = findCatalogPartitions(partitionFilter);
+ List, Path>> selected =
+ filterPartitions(listing.partitionPaths, partitionFilter);
+ List entries = toPartitionEntries(listing.partitions, partitionFilter);
+ return new ScanPlan(enumeratePartitions(selected), rowCount(entries));
}
private List enumeratePartitions(
- List catalogPartitions, @Nullable PartitionPredicate partitionFilter)
- throws IOException {
- List, Path>> partitions =
- toSpecsAndPaths(
- catalogPartitions, coreOptions.formatTablePartitionOnlyValueInPath());
+ List, Path>> partitions) throws IOException {
List splits = new ArrayList<>();
if (partitions.isEmpty()) {
return splits;
}
- FileIO fileIO = table.fileIO();
- // Establish the filesystem on the caller thread so listing workers reuse it under the
- // caller's security context instead of creating it lazily under a shared worker.
- fileIO.exists(new Path(table.location()));
+ FormatTableFileIOResolver fileIOResolver = new FormatTableFileIOResolver(table);
+ boolean tableFileIOPrepared = false;
+ for (Pair, Path> partition : partitions) {
+ boolean useCatalogContextFileIO =
+ fileIOResolver.useCatalogContextFileIO(partition.getValue());
+ if (useCatalogContextFileIO) {
+ fileIOResolver.prepare(partition.getValue(), true);
+ } else if (!tableFileIOPrepared) {
+ fileIOResolver.prepare(partition.getValue(), false);
+ tableFileIOPrepared = true;
+ }
+ }
Function, Path>, List> lister =
pair -> {
BinaryRow partitionRow = toPartitionRow(pair.getKey());
- if (partitionFilter != null && !partitionFilter.test(partitionRow)) {
- return Collections.emptyList();
- }
try {
- return createSplits(fileIO, pair.getValue(), partitionRow);
+ boolean useCatalogContextFileIO =
+ fileIOResolver.useCatalogContextFileIO(pair.getValue());
+ return createSplits(
+ fileIOResolver.fileIO(useCatalogContextFileIO),
+ pair.getValue(),
+ partitionRow,
+ useCatalogContextFileIO);
} catch (FileNotFoundException e) {
warnMissingPartition(pair.getKey(), pair.getValue());
return Collections.emptyList();
@@ -146,12 +155,12 @@ private List enumeratePartitions(
@Override
List, Path>> findPartitions(
@Nullable PartitionPredicate partitionFilter) {
- return toSpecsAndPaths(
- findCatalogPartitions(partitionFilter),
- coreOptions.formatTablePartitionOnlyValueInPath());
+ CatalogPartitionListing listing = findCatalogPartitions(partitionFilter);
+ return filterPartitions(listing.partitionPaths, partitionFilter);
}
- private List findCatalogPartitions(@Nullable PartitionPredicate partitionFilter) {
+ private CatalogPartitionListing findCatalogPartitions(
+ @Nullable PartitionPredicate partitionFilter) {
Optional extracted = FormatTableScan.extractPartitionPredicate(partitionFilter);
Map prefix = leadingEqualityPrefix(extracted);
Predicate catalogFilter = extracted.orElse(null);
@@ -159,7 +168,9 @@ private List findCatalogPartitions(@Nullable PartitionPredicate parti
if (partitions.isEmpty() && prefix.isEmpty() && catalogFilter == null) {
warnIfFilesystemPartitionsExist();
}
- return partitions;
+ List, Path>> partitionPaths =
+ toSpecsAndPaths(partitions, coreOptions.formatTablePartitionOnlyValueInPath());
+ return new CatalogPartitionListing(partitions, partitionPaths);
}
@Override
@@ -169,7 +180,36 @@ List listPartitionEntries() {
@Override
List listPartitionEntries(@Nullable PartitionPredicate partitionFilter) {
- return toPartitionEntries(findCatalogPartitions(partitionFilter), partitionFilter);
+ return toPartitionEntries(
+ findCatalogPartitions(partitionFilter).partitions, partitionFilter);
+ }
+
+ private static final class CatalogPartitionListing {
+
+ private final List partitions;
+ private final List, Path>> partitionPaths;
+
+ private CatalogPartitionListing(
+ List partitions,
+ List, Path>> partitionPaths) {
+ this.partitions = partitions;
+ this.partitionPaths = partitionPaths;
+ }
+ }
+
+ private List, Path>> filterPartitions(
+ List, Path>> partitions,
+ @Nullable PartitionPredicate partitionFilter) {
+ if (partitionFilter == null) {
+ return partitions;
+ }
+ List, Path>> selected = new ArrayList<>();
+ for (Pair, Path> partition : partitions) {
+ if (partitionFilter.test(toPartitionRow(partition.getKey()))) {
+ selected.add(partition);
+ }
+ }
+ return selected;
}
private List toPartitionEntries(
@@ -219,16 +259,39 @@ private OptionalLong rowCount(List entries) {
private List, Path>> toSpecsAndPaths(
List partitions, boolean onlyValueInPath) {
+ if (partitions.stream().noneMatch(partition -> partition.location() != null)) {
+ return toDefaultSpecsAndPaths(partitions, onlyValueInPath);
+ }
+ List, Path>> result = new ArrayList<>(partitions.size());
+ Path tablePath = new Path(table.location());
+ FormatTablePartitionPathResolver pathResolver =
+ new FormatTablePartitionPathResolver(
+ tablePath, table.fullName(), onlyValueInPath, table.catalogContext());
+ for (Partition partition : partitions) {
+ LinkedHashMap spec = normalizeSpec(partition.spec(), onlyValueInPath);
+ Path partitionPath = pathResolver.resolve(spec, partition.location());
+ if (pathResolver.validateAndRecord(spec, partitionPath)) {
+ result.add(Pair.of(spec, partitionPath));
+ }
+ }
+ return result;
+ }
+
+ private List, Path>> toDefaultSpecsAndPaths(
+ List partitions, boolean onlyValueInPath) {
List, Path>> result = new ArrayList<>(partitions.size());
+ Set> seen = new HashSet<>(partitions.size());
Path tablePath = new Path(table.location());
- // A duplicate catalog entry must not duplicate all records in that partition.
- Set seenPartitionPaths = new HashSet<>(partitions.size());
for (Partition partition : partitions) {
LinkedHashMap spec = normalizeSpec(partition.spec(), onlyValueInPath);
- String partitionPath =
- PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath);
- if (seenPartitionPaths.add(partitionPath)) {
- result.add(Pair.of(spec, new Path(tablePath, partitionPath)));
+ if (seen.add(spec)) {
+ result.add(
+ Pair.of(
+ spec,
+ new Path(
+ tablePath,
+ PartitionPathUtils.generatePartitionPathUtil(
+ spec, onlyValueInPath))));
}
}
return result;
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
index 78f2f9532e8a..614780e42521 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatDataSplit.java
@@ -39,10 +39,17 @@ public class FormatDataSplit implements Split {
private final List files;
@Nullable private final BinaryRow partition;
+ private final boolean useCatalogContextFileIO;
public FormatDataSplit(List files, @Nullable BinaryRow partition) {
+ this(files, partition, false);
+ }
+
+ public FormatDataSplit(
+ List files, @Nullable BinaryRow partition, boolean useCatalogContextFileIO) {
this.files = files;
this.partition = partition;
+ this.useCatalogContextFileIO = useCatalogContextFileIO;
}
public List files() {
@@ -54,6 +61,14 @@ public BinaryRow partition() {
return partition;
}
+ /**
+ * Whether readers must resolve this split through the client {@code CatalogContext} instead of
+ * the FileIO bound to the table root.
+ */
+ public boolean useCatalogContextFileIO() {
+ return useCatalogContextFileIO;
+ }
+
/** Total bytes to read for this split, i.e. the sum of {@link FileMeta#readSize()}. */
public long totalSize() {
return files.stream().mapToLong(FileMeta::readSize).sum();
@@ -83,12 +98,14 @@ public boolean equals(Object o) {
return false;
}
FormatDataSplit that = (FormatDataSplit) o;
- return Objects.equals(files, that.files) && Objects.equals(partition, that.partition);
+ return useCatalogContextFileIO == that.useCatalogContextFileIO
+ && Objects.equals(files, that.files)
+ && Objects.equals(partition, that.partition);
}
@Override
public int hashCode() {
- return Objects.hash(files, partition);
+ return Objects.hash(files, partition, useCatalogContextFileIO);
}
/**
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
index ff5d73e48f1e..924bb3c9dc25 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
@@ -25,6 +25,7 @@
import org.apache.paimon.format.FileFormatDiscover;
import org.apache.paimon.format.FormatReaderContext;
import org.apache.paimon.format.FormatReaderFactory;
+import org.apache.paimon.fs.FileIO;
import org.apache.paimon.io.DataFileRecordReader;
import org.apache.paimon.mergetree.compact.ConcatRecordReader;
import org.apache.paimon.options.CatalogOptions;
@@ -78,6 +79,7 @@ public class FormatReadBuilder implements ReadBuilder {
@Nullable private Predicate filter;
@Nullable private PartitionPredicate partitionFilter;
@Nullable private Integer limit;
+ @Nullable private transient FormatTableFileIOResolver fileIOResolver;
public FormatReadBuilder(FormatTable table) {
this.table = table;
@@ -204,11 +206,13 @@ protected RecordReader createReader(
table.partitionKeys(), readType().getFields(), table.partitionType());
BinaryRow partition = dataSplit.partition();
+ FileIO fileIO = fileIOResolver().fileIO(dataSplit.useCatalogContextFileIO());
List> suppliers = new ArrayList<>();
for (FormatDataSplit.FileMeta file : dataSplit.files()) {
suppliers.add(
() ->
createFileReader(
+ fileIO,
file,
partition,
readerFactory,
@@ -219,6 +223,7 @@ protected RecordReader createReader(
}
private RecordReader createFileReader(
+ FileIO fileIO,
FormatDataSplit.FileMeta file,
@Nullable BinaryRow partition,
FormatReaderFactory readerFactory,
@@ -227,7 +232,7 @@ private RecordReader createFileReader(
throws IOException {
FormatReaderContext formatReaderContext =
new FormatReaderContext(
- table.fileIO(), file.filePath(), file.fileSize(), null, readBatchSizer);
+ fileIO, file.filePath(), file.fileSize(), null, readBatchSizer);
try {
FileRecordReader reader;
Long length = file.length();
@@ -271,6 +276,13 @@ private static RowType getRowTypeWithoutPartition(RowType rowType, List
.collect(Collectors.toList()));
}
+ private synchronized FormatTableFileIOResolver fileIOResolver() {
+ if (fileIOResolver == null) {
+ fileIOResolver = new FormatTableFileIOResolver(table);
+ }
+ return fileIOResolver;
+ }
+
// ===================== Unsupported ===============================
@Override
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 76de97dcfb39..72255931277f 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
@@ -63,12 +63,14 @@
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
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;
@@ -94,6 +96,7 @@ public class FormatTableCommit implements BatchTableCommit {
protected boolean overwrite = false;
private Catalog hiveCatalog;
private Identifier tableIdentifier;
+ private final CatalogContext catalogContext;
@Nullable private final FormatTablePartitionManager partitionManager;
private final boolean dynamicPartitionOverwrite;
private final int cleanupThreadNum;
@@ -165,6 +168,7 @@ public FormatTableCommit(
this.overwrite = overwrite;
this.partitionKeys = partitionKeys;
this.tableIdentifier = tableIdentifier;
+ this.catalogContext = catalogContext;
this.partitionManager = partitionManager;
this.dynamicPartitionOverwrite = dynamicPartitionOverwrite;
this.cleanupThreadNum = cleanupThreadNum;
@@ -203,6 +207,8 @@ public void commit(List commitMessages) {
}
}
+ List validatedPartitions = rejectWritesToCustomLocationPartitions(messages);
+
Set> partitionSpecs = new HashSet<>();
Set clearedPartitionPaths = new HashSet<>();
Path staticPartitionPath = null;
@@ -243,7 +249,10 @@ public void commit(List commitMessages) {
// 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(), 0, cleanupThreadNum));
+ deletePreviousDataFiles(
+ tableDataDirectories(validatedPartitions),
+ 0,
+ cleanupThreadNum));
}
}
if (overwrite) {
@@ -388,6 +397,134 @@ private void publishMessages(List messages) throws IOExce
}
}
+ /** Rejects writes whose files would belong to a catalog partition outside the table root. */
+ private List rejectWritesToCustomLocationPartitions(
+ List messages) {
+ if (partitionManager == null || partitionKeys == null || partitionKeys.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ try {
+ return rejectWritesToCustomLocationPartitionsBeforeMutation(messages);
+ } 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.
+ markPublishedTargetsToPreserveOnAbort(messages);
+ throw failure;
+ }
+ }
+
+ private List rejectWritesToCustomLocationPartitionsBeforeMutation(
+ List messages) {
+ Predicate affectsPartition;
+ boolean hasStaticPrefixWithoutFiles = false;
+ if (overwrite && staticPartitions != null && !staticPartitions.isEmpty()) {
+ LinkedHashMap staticSpec = orderedPartitionPrefix(staticPartitions);
+ if (staticSpec.size() == partitionKeys.size()) {
+ affectsPartition = partition -> partition.spec().equals(staticPartitions);
+ } else {
+ affectsPartition =
+ partition -> partitionSpecMatchesPrefix(partition.spec(), staticSpec);
+ }
+ } else if (overwrite && !replacesOnlyWrittenPartitions()) {
+ affectsPartition = ignored -> true;
+ } else {
+ Set> affectedSpecs = new LinkedHashSet<>();
+ for (TwoPhaseCommitMessage message : messages) {
+ Path targetPath = message.getCommitter().targetPath();
+ if (targetPath == null) {
+ // Preserve the established failure order for a malformed committer. The
+ // publish or registration path will report its own contract violation.
+ continue;
+ }
+ affectedSpecs.add(
+ extractPartitionSpecFromPath(targetPath.getParent(), partitionKeys));
+ }
+ if (!overwrite && staticPartitions != null && !staticPartitions.isEmpty()) {
+ LinkedHashMap staticSpec = orderedPartitionPrefix(staticPartitions);
+ if (staticSpec.size() == partitionKeys.size()) {
+ if (affectedSpecs.isEmpty()) {
+ affectedSpecs.add(staticPartitions);
+ }
+ } else {
+ hasStaticPrefixWithoutFiles = affectedSpecs.isEmpty();
+ }
+ }
+ if (affectedSpecs.isEmpty() && !hasStaticPrefixWithoutFiles) {
+ return Collections.emptyList();
+ }
+ affectsPartition =
+ affectedSpecs.isEmpty()
+ ? ignored -> false
+ : partition -> affectedSpecs.contains(partition.spec());
+ }
+
+ List registry = loadPartitionRegistry();
+ List affectedPartitions =
+ registry.stream().filter(affectsPartition).collect(Collectors.toList());
+ for (Partition partition : affectedPartitions) {
+ if (partition.location() != null) {
+ throw unsupportedCustomLocation(overwrite ? "Overwriting" : "Writing", partition);
+ }
+ }
+ return registry;
+ }
+
+ private LinkedHashMap orderedPartitionPrefix(
+ Map partitionSpec) {
+ if (partitionSpec.size() > partitionKeys.size()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition spec %s is not a leading prefix of partition keys %s.",
+ partitionSpec, partitionKeys));
+ }
+ LinkedHashMap orderedSpec = new LinkedHashMap<>();
+ for (int i = 0; i < partitionSpec.size(); i++) {
+ String key = partitionKeys.get(i);
+ if (!partitionSpec.containsKey(key)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition spec %s is not a leading prefix of partition keys %s.",
+ partitionSpec, partitionKeys));
+ }
+ orderedSpec.put(key, partitionSpec.get(key));
+ }
+ return orderedSpec;
+ }
+
+ /** Validates every registered path before using it for a write or truncate decision. */
+ private List loadPartitionRegistry() {
+ List partitions = partitionManager.listPartitions(Collections.emptyMap(), null);
+ FormatTablePartitionRegistryValidator.validatePartitionLocations(
+ partitions,
+ partitionKeys,
+ new Path(location),
+ tableIdentifier.getFullName(),
+ formatTablePartitionOnlyValueInPath,
+ catalogContext);
+ return partitions;
+ }
+
+ private static boolean partitionSpecMatchesPrefix(
+ Map partitionSpec, Map prefix) {
+ for (Map.Entry entry : prefix.entrySet()) {
+ if (!partitionSpec.containsKey(entry.getKey())
+ || !Objects.equals(entry.getValue(), partitionSpec.get(entry.getKey()))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private UnsupportedOperationException unsupportedCustomLocation(
+ String operation, Partition partition) {
+ return new UnsupportedOperationException(
+ String.format(
+ "%s catalog-managed Format Table partition %s with custom location "
+ + "'%s' is not supported.",
+ operation, partition.spec(), partition.location()));
+ }
+
private List publishMessage(TwoPhaseCommitMessage message) {
try {
message.getCommitter().commit(fileIO);
@@ -436,7 +573,8 @@ private void reportPartitions(
new ArrayList<>(specs),
true,
new ArrayList<>(statisticsByPartition.values()),
- replaceStatistics);
+ replaceStatistics,
+ null);
}
/** What one commit wrote into a partition, with one more of its files folded in. */
@@ -674,17 +812,17 @@ private boolean replacesOnlyWrittenPartitions() {
* 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.
*/
- private List tableDataDirectories() {
+ private List tableDataDirectories(List validatedPartitions) {
if (partitionKeys == null || partitionKeys.isEmpty()) {
return Collections.singletonList(new Path(location));
}
List directories = new ArrayList<>();
if (partitionManager != null) {
- for (Map spec : registeredPartitions(Collections.emptyMap())) {
+ for (Partition partition : validatedPartitions) {
directories.add(
buildPartitionPath(
location,
- spec,
+ partition.spec(),
formatTablePartitionOnlyValueInPath,
partitionKeys));
}
@@ -1070,7 +1208,13 @@ public void truncateTable() {
// Emptying the table is emptying every partition it has, and which those are is answered
// by whatever the table reads its partitions from.
if (partitionManager != null) {
- truncate(registeredPartitions(Collections.emptyMap()));
+ List partitions = loadPartitionRegistry();
+ for (Partition partition : partitions) {
+ if (partition.location() != null) {
+ throw unsupportedCustomLocation("Truncating", partition);
+ }
+ }
+ truncate(partitions.stream().map(Partition::spec).collect(Collectors.toList()));
return;
}
// Filesystem partition discovery: the partition directories the scan reads are the table.
@@ -1091,45 +1235,47 @@ public void truncateTable() {
@Override
public void truncatePartitions(List> partitionSpecs) {
- if (partitionManager == null) {
- truncate(partitionSpecs);
+ if (partitionSpecs.isEmpty()) {
return;
}
- // Complete specs are asked for in one request; only a prefix has to be listed on its own.
- List> complete = new ArrayList<>();
+ List> normalizedSpecs = new ArrayList<>(partitionSpecs.size());
for (Map partitionSpec : partitionSpecs) {
- if (partitionSpec.size() == partitionKeys.size()) {
- complete.add(partitionSpec);
- }
+ normalizedSpecs.add(orderedPartitionPrefix(partitionSpec));
}
- Set> registered =
- complete.isEmpty()
- ? Collections.emptySet()
- : partitionManager.listPartitionsByNames(complete).stream()
- .map(Partition::spec)
- .collect(Collectors.toSet());
- List> partitions = new ArrayList<>();
- for (Map partitionSpec : partitionSpecs) {
- if (partitionSpec.size() == partitionKeys.size()) {
- if (registered.contains(partitionSpec)) {
- partitions.add(partitionSpec);
- }
- } else {
- partitions.addAll(registeredPartitions(partitionSpec));
+ if (partitionManager == null) {
+ truncate(normalizedSpecs);
+ return;
+ }
+ List registry = loadPartitionRegistry();
+ Map, Partition> partitions =
+ selectRequestedPartitions(registry, normalizedSpecs);
+ for (Partition partition : partitions.values()) {
+ if (partition.location() != null) {
+ throw unsupportedCustomLocation("Truncating", partition);
}
}
- truncate(partitions);
+ truncate(partitions.values().stream().map(Partition::spec).collect(Collectors.toList()));
}
- /**
- * The registered partitions named by {@code prefix}, which names only the leading partition
- * keys, or none of them. The catalog says which partitions a catalog-managed table has, so
- * truncating neither empties nor registers a directory still waiting for MSCK REPAIR TABLE.
- */
- private List> registeredPartitions(Map prefix) {
- return partitionManager.listPartitions(prefix, null).stream()
- .map(Partition::spec)
- .collect(Collectors.toList());
+ private Map, Partition> selectRequestedPartitions(
+ List registry, List> partitionSpecs) {
+ Map, Partition> selected = new LinkedHashMap<>();
+ Set> requestedPrefixes = new HashSet<>(partitionSpecs);
+ for (Partition partition : registry) {
+ if (requestedPrefixes.contains(Collections.emptyMap())) {
+ selected.putIfAbsent(partition.spec(), partition);
+ continue;
+ }
+ LinkedHashMap registeredPrefix = new LinkedHashMap<>();
+ for (String partitionKey : partitionKeys) {
+ registeredPrefix.put(partitionKey, partition.spec().get(partitionKey));
+ if (requestedPrefixes.contains(registeredPrefix)) {
+ selected.putIfAbsent(partition.spec(), partition);
+ break;
+ }
+ }
+ }
+ return selected;
}
private void truncate(List> partitionSpecs) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileIOResolver.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileIOResolver.java
new file mode 100644
index 000000000000..ef614b58d986
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableFileIOResolver.java
@@ -0,0 +1,87 @@
+/*
+ * 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.catalog.CatalogContext;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.ResolvingFileIO;
+import org.apache.paimon.table.FormatTable;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+
+/**
+ * Uses the table FileIO under the table root and the catalog-context FileIO outside it. The choice
+ * comes from the registered partition path, not a listed file URI.
+ */
+final class FormatTableFileIOResolver {
+
+ private final Path tableRoot;
+ private final FileIO tableFileIO;
+ @Nullable private final CatalogContext catalogContext;
+ @Nullable private transient volatile ResolvingFileIO catalogContextFileIO;
+
+ FormatTableFileIOResolver(FormatTable table) {
+ this.tableRoot = new Path(table.location());
+ this.tableFileIO = table.fileIO();
+ this.catalogContext = table.catalogContext();
+ }
+
+ boolean useCatalogContextFileIO(Path partitionPath) {
+ return !FormatTablePartitionPathResolver.isWithin(partitionPath, tableRoot, catalogContext);
+ }
+
+ /**
+ * Resolves an external filesystem on the caller thread before parallel listing starts. The
+ * underlying resolver caches the result by scheme and authority.
+ */
+ void prepare(Path path, boolean useCatalogContextFileIO) throws IOException {
+ if (useCatalogContextFileIO) {
+ catalogContextFileIO().fileIO(path);
+ } else {
+ tableFileIO.exists(tableRoot);
+ }
+ }
+
+ FileIO fileIO(boolean useCatalogContextFileIO) {
+ return useCatalogContextFileIO ? catalogContextFileIO() : tableFileIO;
+ }
+
+ private ResolvingFileIO catalogContextFileIO() {
+ ResolvingFileIO result = catalogContextFileIO;
+ if (result != null) {
+ return result;
+ }
+ synchronized (this) {
+ result = catalogContextFileIO;
+ if (result == null) {
+ if (catalogContext == null) {
+ throw new IllegalStateException(
+ "A CatalogContext is required to access a Format Table partition outside the table root.");
+ }
+ result = new ResolvingFileIO();
+ result.configure(catalogContext);
+ catalogContextFileIO = result;
+ }
+ return result;
+ }
+ }
+}
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 aaacebbe7159..d14238b7d0f1 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
@@ -66,29 +66,16 @@ public interface FormatTablePartitionManager extends Serializable {
* whole batch is rejected when any partition already exists, so such a request is never split.
*/
default void createPartitions(List> partitions, boolean ignoreIfExists) {
- createPartitions(partitions, ignoreIfExists, null, false);
+ createPartitions(partitions, ignoreIfExists, null, false, null);
}
- /**
- * Register partitions and report statistics for them in the same call, so a partition is never
- * registered by a request whose statistics failed on their own.
- *
- * Statistics are matched to {@code partitions} by {@link PartitionStatistics#spec()} and may
- * cover only some of them; {@code replaceStatistics} says whether they replace what the catalog
- * 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.
- *
- *
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
- * report and leave the caller no way to notice.
- */
+ /** Register partitions with optional statistics and position-aligned locations. */
void createPartitions(
List> partitions,
boolean ignoreIfExists,
@Nullable List statistics,
- boolean replaceStatistics);
+ boolean replaceStatistics,
+ @Nullable List partitionLocations);
/** Unregister partitions. Metadata only; missing partitions are ignored. */
void dropPartitions(List> partitions);
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
new file mode 100644
index 000000000000..c7822f804c8a
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionPathResolver.java
@@ -0,0 +1,492 @@
+/*
+ * 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.catalog.CatalogContext;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.utils.PartitionPathUtils;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.net.NetUtils;
+
+import javax.annotation.Nullable;
+
+import java.io.ByteArrayOutputStream;
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+
+/** Resolves catalog-managed Format Table partition paths and rejects ambiguous catalog metadata. */
+public final class FormatTablePartitionPathResolver {
+
+ private final Path tablePath;
+ private final String tableName;
+ private final boolean onlyValueInPath;
+ @Nullable private final CatalogContext catalogContext;
+ private final Map, String> pathsBySpec = new LinkedHashMap<>();
+ private final Map ownershipRoots = new HashMap<>();
+
+ FormatTablePartitionPathResolver(Path tablePath, String tableName, boolean onlyValueInPath) {
+ this(tablePath, tableName, onlyValueInPath, null);
+ }
+
+ FormatTablePartitionPathResolver(
+ Path tablePath,
+ String tableName,
+ boolean onlyValueInPath,
+ @Nullable CatalogContext catalogContext) {
+ this.tablePath = tablePath;
+ this.tableName = tableName;
+ this.onlyValueInPath = onlyValueInPath;
+ this.catalogContext = catalogContext;
+ }
+
+ Path resolve(LinkedHashMap spec, @Nullable String customLocation) {
+ Path defaultPath =
+ new Path(
+ tablePath,
+ PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath));
+ if (customLocation == null) {
+ return defaultPath;
+ }
+
+ try {
+ return resolveCustomLocation(
+ tablePath, spec, onlyValueInPath, customLocation, catalogContext);
+ } catch (IllegalArgumentException e) {
+ throw invalidLocation(spec, e);
+ }
+ }
+
+ /** Resolves a custom location using the catalog's Hadoop filesystem identity. */
+ public static Path resolveCustomLocation(
+ Path tablePath,
+ LinkedHashMap spec,
+ boolean onlyValueInPath,
+ String customLocation,
+ @Nullable CatalogContext catalogContext) {
+ PartitionPathUtils.validatePartitionSpecForPath(spec, onlyValueInPath);
+ Path customPath = canonicalizeCustomLocation(customLocation, catalogContext);
+ if (usesViewFileSystem(tablePath) || usesViewFileSystem(customPath)) {
+ throw new IllegalArgumentException(
+ "Custom ViewFS partition locations require mount-table identity resolution.");
+ }
+ if (overlaps(customPath, tablePath, catalogContext)) {
+ throw new IllegalArgumentException("Custom partition location overlaps table data.");
+ }
+ return customPath;
+ }
+
+ /**
+ * Records a resolved path. Returns false for a repeated identical spec and path; callers skip
+ * that entry so duplicate catalog rows do not produce duplicate data.
+ */
+ boolean validateAndRecord(LinkedHashMap spec, Path path) {
+ ResolvedPath resolved = ResolvedPath.of(path, catalogContext);
+ String previousForSpec = pathsBySpec.get(spec);
+ if (previousForSpec != null) {
+ if (previousForSpec.equals(path.toString())) {
+ return false;
+ }
+ throw overlappingLocations();
+ }
+
+ if (overlapsOwnedPath(resolved)) {
+ throw overlappingLocations();
+ }
+ pathsBySpec.put(new LinkedHashMap<>(spec), path.toString());
+ return true;
+ }
+
+ private boolean overlapsOwnedPath(ResolvedPath path) {
+ OwnershipNode node =
+ ownershipRoots.computeIfAbsent(path.fileSystem, ignored -> new OwnershipNode());
+ String[] segments = path.pathSegments();
+ for (String segment : segments) {
+ // A terminal node reached before the candidate ends is an existing ancestor.
+ if (node.owned) {
+ return true;
+ }
+ node = node.children.computeIfAbsent(segment, ignored -> new OwnershipNode());
+ }
+ // A terminal final node is equality. Children below it make the candidate an ancestor.
+ if (node.owned || !node.children.isEmpty()) {
+ return true;
+ }
+ node.owned = true;
+ return false;
+ }
+
+ /** Canonicalizes a custom location using the catalog's Hadoop configuration when present. */
+ public static Path canonicalizeCustomLocation(
+ 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);
+ URI uri = path.toUri();
+ String scheme = uri.getScheme();
+ String authority = uri.getAuthority();
+ String uriPath = uri.getPath();
+ if (scheme == null
+ || scheme.isEmpty()
+ || (uri.getUserInfo() != null && !isAbfsAuthority(uri))
+ || uriPath == null
+ || !uriPath.startsWith(Path.SEPARATOR)
+ || uriPath.equals(Path.SEPARATOR)) {
+ throw new IllegalArgumentException("Invalid custom partition location.");
+ }
+
+ 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
+ : authority.toLowerCase(Locale.ROOT);
+ Path canonical = new Path(scheme, authority, uriPath);
+ return scheme.equals("hdfs")
+ ? canonicalizeHdfsPath(canonical, catalogContext)
+ : canonical;
+ } catch (IllegalArgumentException e) {
+ throw e;
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException("Invalid custom partition location.", e);
+ }
+ }
+
+ private static boolean isAbfsAuthority(URI uri) {
+ String scheme = uri.getScheme();
+ String userInfo = uri.getUserInfo();
+ return scheme != null
+ && (scheme.equalsIgnoreCase("abfs") || scheme.equalsIgnoreCase("abfss"))
+ && userInfo != null
+ && !userInfo.isEmpty()
+ && userInfo.indexOf(':') < 0
+ && uri.getHost() != null;
+ }
+
+ private static boolean usesViewFileSystem(Path path) {
+ String scheme = path.toUri().getScheme();
+ return scheme != null && scheme.equalsIgnoreCase("viewfs");
+ }
+
+ private static Path canonicalizeHdfsPath(Path path, @Nullable CatalogContext catalogContext) {
+ URI canonicalUri = canonicalHdfsUri(path.toUri(), catalogContext);
+ if (canonicalUri.getAuthority() == null) {
+ throw new IllegalArgumentException(
+ "Authorityless HDFS location requires an HDFS default filesystem.");
+ }
+ return new Path("hdfs", canonicalUri.getAuthority(), path.toUri().getPath());
+ }
+
+ private static URI canonicalHdfsUri(URI uri, @Nullable CatalogContext catalogContext) {
+ URI resolved = uri;
+ if (resolved.getAuthority() == null && catalogContext != null) {
+ URI defaultUri = FileSystem.getDefaultUri(catalogContext.hadoopConf());
+ if ("hdfs".equalsIgnoreCase(defaultUri.getScheme())
+ && defaultUri.getAuthority() != null) {
+ resolved = defaultUri;
+ }
+ }
+ if (resolved.getAuthority() == null) {
+ return resolved;
+ }
+ String logicalNameservice = logicalHdfsNameservice(resolved, catalogContext);
+ if (logicalNameservice != null) {
+ return new Path("hdfs", logicalNameservice, "/").toUri();
+ }
+ URI physical = NetUtils.getCanonicalUri(resolved, 8020);
+ return new Path("hdfs", physical.getAuthority().toLowerCase(Locale.ROOT), Path.SEPARATOR)
+ .toUri();
+ }
+
+ @Nullable
+ private static String logicalHdfsNameservice(URI uri, @Nullable CatalogContext catalogContext) {
+ if (catalogContext == null || uri.getHost() == null) {
+ return null;
+ }
+ String nameservices = catalogContext.hadoopConf().get("dfs.nameservices", "");
+ String requestedAuthority = canonicalHdfsAuthority(uri);
+ String match = null;
+ for (String name : nameservices.split(",")) {
+ String nameservice = name.trim();
+ if (nameservice.isEmpty()) {
+ continue;
+ }
+ boolean matches =
+ nameservice.equalsIgnoreCase(uri.getHost())
+ || matchesConfiguredHdfsAddress(
+ requestedAuthority,
+ catalogContext
+ .hadoopConf()
+ .get("dfs.namenode.rpc-address." + nameservice));
+ String namenodes =
+ catalogContext.hadoopConf().get("dfs.ha.namenodes." + nameservice, "");
+ for (String node : namenodes.split(",")) {
+ String nodeId = node.trim();
+ if (nodeId.isEmpty()) {
+ continue;
+ }
+ String address =
+ catalogContext
+ .hadoopConf()
+ .get("dfs.namenode.rpc-address." + nameservice + "." + nodeId);
+ if (address == null || address.trim().isEmpty()) {
+ continue;
+ }
+ if (matchesConfiguredHdfsAddress(requestedAuthority, address)) {
+ matches = true;
+ }
+ }
+ if (matches) {
+ if (match != null && !match.equals(nameservice)) {
+ throw new IllegalArgumentException(
+ "HDFS authority belongs to multiple logical nameservices.");
+ }
+ match = nameservice;
+ }
+ }
+ return match;
+ }
+
+ private static boolean matchesConfiguredHdfsAddress(
+ String requestedAuthority, @Nullable String address) {
+ if (address == null || address.trim().isEmpty()) {
+ return false;
+ }
+ URI member = URI.create("hdfs://" + address.trim());
+ return requestedAuthority.equals(canonicalHdfsAuthority(member));
+ }
+
+ private static String canonicalHdfsAuthority(URI uri) {
+ return NetUtils.getCanonicalUri(uri, 8020).getAuthority().toLowerCase(Locale.ROOT);
+ }
+
+ private static void validateDecodedLocation(String location) {
+ if (location == null
+ || location.isEmpty()
+ || isBoundaryWhitespace(location)
+ || location.contains("?")
+ || location.contains("#")
+ || location.contains("\\")) {
+ throw new IllegalArgumentException("Invalid custom partition location.");
+ }
+
+ for (int offset = 0; offset < location.length(); ) {
+ int codePoint = location.codePointAt(offset);
+ if (Character.isISOControl(codePoint)) {
+ throw new IllegalArgumentException("Invalid custom partition location.");
+ }
+ offset += Character.charCount(codePoint);
+ }
+
+ for (String segment : location.split(Path.SEPARATOR, -1)) {
+ if (segment.equals(Path.CUR_DIR) || segment.equals("..")) {
+ throw new IllegalArgumentException("Invalid custom partition location.");
+ }
+ }
+ }
+
+ private static boolean isBoundaryWhitespace(String value) {
+ int first = value.codePointAt(0);
+ int last = value.codePointBefore(value.length());
+ return isWhitespace(first) || isWhitespace(last);
+ }
+
+ private static boolean isWhitespace(int codePoint) {
+ return Character.isWhitespace(codePoint) || Character.isSpaceChar(codePoint);
+ }
+
+ private static String decodePercentOnce(String value) {
+ StringBuilder decoded = new StringBuilder(value.length());
+ for (int offset = 0; offset < value.length(); ) {
+ char current = value.charAt(offset);
+ if (current != '%') {
+ decoded.append(current);
+ offset++;
+ continue;
+ }
+
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ while (offset < value.length() && value.charAt(offset) == '%') {
+ if (offset + 2 >= value.length()) {
+ throw new IllegalArgumentException("Invalid percent encoding in location.");
+ }
+ int high = Character.digit(value.charAt(offset + 1), 16);
+ int low = Character.digit(value.charAt(offset + 2), 16);
+ if (high < 0 || low < 0) {
+ throw new IllegalArgumentException("Invalid percent encoding in location.");
+ }
+ bytes.write((high << 4) + low);
+ offset += 3;
+ }
+ try {
+ decoded.append(
+ StandardCharsets.UTF_8
+ .newDecoder()
+ .onMalformedInput(CodingErrorAction.REPORT)
+ .onUnmappableCharacter(CodingErrorAction.REPORT)
+ .decode(ByteBuffer.wrap(bytes.toByteArray())));
+ } catch (CharacterCodingException e) {
+ throw new IllegalArgumentException("Invalid percent encoding in location.", e);
+ }
+ }
+ return decoded.toString();
+ }
+
+ static boolean isWithin(Path candidate, Path root) {
+ return isWithin(candidate, root, null);
+ }
+
+ static boolean isWithin(Path candidate, Path root, @Nullable CatalogContext catalogContext) {
+ ResolvedPath candidatePath = ResolvedPath.of(candidate, catalogContext);
+ ResolvedPath rootPath = ResolvedPath.of(root, catalogContext);
+ return rootPath.equals(candidatePath) || rootPath.isAncestorOf(candidatePath);
+ }
+
+ private static boolean overlaps(
+ Path left, Path right, @Nullable CatalogContext catalogContext) {
+ ResolvedPath resolvedLeft = ResolvedPath.of(left, catalogContext);
+ ResolvedPath resolvedRight = ResolvedPath.of(right, catalogContext);
+ return resolvedLeft.equals(resolvedRight)
+ || resolvedLeft.isAncestorOf(resolvedRight)
+ || resolvedRight.isAncestorOf(resolvedLeft);
+ }
+
+ private IllegalStateException invalidLocation(
+ Map spec, IllegalArgumentException cause) {
+ return new IllegalStateException(
+ String.format(
+ "Catalog returned an invalid custom location for partition %s of Format Table %s.",
+ spec, tableName),
+ cause);
+ }
+
+ private IllegalStateException overlappingLocations() {
+ return new IllegalStateException(
+ String.format(
+ "Catalog returned overlapping locations for different partitions of Format Table %s.",
+ tableName));
+ }
+
+ /**
+ * One trie is maintained per filesystem. Visiting each path segment once is sufficient:
+ * ancestors are terminal nodes on the route, equality is the terminal node at the route's end,
+ * and descendants are children below that node.
+ */
+ private static final class OwnershipNode {
+
+ private final Map children = new HashMap<>();
+ private boolean owned;
+ }
+
+ private static final class ResolvedPath {
+
+ private final String fileSystem;
+ private final String path;
+
+ private ResolvedPath(String fileSystem, String path) {
+ this.fileSystem = fileSystem;
+ this.path = path;
+ }
+
+ private static ResolvedPath of(Path path, @Nullable CatalogContext catalogContext) {
+ URI uri = path.toUri().normalize();
+ String scheme = canonicalFileSystemScheme(uri.getScheme());
+ URI fileSystemUri = scheme.equals("hdfs") ? canonicalHdfsUri(uri, catalogContext) : uri;
+ String authority = fileSystemUri.getAuthority();
+ authority = authority == null ? "" : authority.toLowerCase(Locale.ROOT);
+ String normalizedPath = trimTrailingSeparators(uri.getPath());
+ return new ResolvedPath(scheme + "://" + authority, normalizedPath);
+ }
+
+ private static String canonicalFileSystemScheme(@Nullable String scheme) {
+ // An absolute path without a scheme and file:/ name the same local filesystem.
+ if (scheme == null) {
+ return "file";
+ }
+ String normalized = scheme.toLowerCase(Locale.ROOT);
+ // These aliases address the same storage namespaces with different clients or
+ // transport settings and therefore cannot establish separate ownership boundaries.
+ if (normalized.equals("abfss")) {
+ return "abfs";
+ }
+ if (normalized.equals("s3a") || normalized.equals("s3n")) {
+ return "s3";
+ }
+ return normalized;
+ }
+
+ private boolean isAncestorOf(ResolvedPath other) {
+ if (!fileSystem.equals(other.fileSystem) || path.equals(other.path)) {
+ return false;
+ }
+ if (path.equals(Path.SEPARATOR)) {
+ return other.path.startsWith(Path.SEPARATOR);
+ }
+ return other.path.startsWith(path + Path.SEPARATOR);
+ }
+
+ private String[] pathSegments() {
+ return path.substring(1).split(Path.SEPARATOR, -1);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ResolvedPath that = (ResolvedPath) o;
+ return fileSystem.equals(that.fileSystem) && path.equals(that.path);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(fileSystem, path);
+ }
+
+ private static String trimTrailingSeparators(String path) {
+ int end = path.length();
+ while (end > 1 && path.charAt(end - 1) == Path.SEPARATOR_CHAR) {
+ end--;
+ }
+ return path.substring(0, end);
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidator.java
new file mode 100644
index 000000000000..acd4ff599754
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidator.java
@@ -0,0 +1,64 @@
+/*
+ * 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.catalog.CatalogContext;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.partition.Partition;
+
+import javax.annotation.Nullable;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Rejects incomplete specs and partition locations that resolve to the same or nested paths. */
+public final class FormatTablePartitionRegistryValidator {
+
+ private FormatTablePartitionRegistryValidator() {}
+
+ public static void validatePartitionLocations(
+ List partitions,
+ List partitionKeys,
+ Path tablePath,
+ String tableName,
+ boolean onlyValueInPath,
+ @Nullable CatalogContext catalogContext) {
+ FormatTablePartitionPathResolver resolver =
+ new FormatTablePartitionPathResolver(
+ tablePath, tableName, onlyValueInPath, catalogContext);
+ for (Partition partition : partitions) {
+ Map spec = partition.spec();
+ if (spec == null
+ || spec.size() != partitionKeys.size()
+ || !spec.keySet().containsAll(partitionKeys)) {
+ throw new IllegalStateException(
+ String.format(
+ "Catalog returned incomplete partition spec %s for Format Table %s.",
+ spec, tableName));
+ }
+ LinkedHashMap orderedSpec = new LinkedHashMap<>();
+ for (String partitionKey : partitionKeys) {
+ orderedSpec.put(partitionKey, spec.get(partitionKey));
+ }
+ Path resolved = resolver.resolve(orderedSpec, partition.location());
+ resolver.validateAndRecord(orderedSpec, resolved);
+ }
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
index c46919576d9d..b6578fd85905 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/SplitEnumerator.java
@@ -145,6 +145,15 @@ BinaryRow toPartitionRow(LinkedHashMap partitionSpec) {
List createSplits(FileIO fileIO, Path path, @Nullable BinaryRow partition)
throws IOException {
+ return createSplits(fileIO, path, partition, false);
+ }
+
+ List createSplits(
+ FileIO fileIO,
+ Path path,
+ @Nullable BinaryRow partition,
+ boolean useCatalogContextFileIO)
+ throws IOException {
List segments = new ArrayList<>();
// The listed directory is a single partition, or the table itself when unpartitioned.
List files = FormatTableScan.listDataFiles(fileIO, path);
@@ -159,7 +168,7 @@ List createSplits(FileIO fileIO, Path path, @Nullable BinaryRow partition
segments,
file -> Math.max(file.readSize(), openFileCost),
targetSplitSize)) {
- splits.add(new FormatDataSplit(bin, partition));
+ splits.add(new FormatDataSplit(bin, partition, useCatalogContextFileIO));
}
return splits;
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
index 37b991fc6846..c94b9e51d13a 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
@@ -79,6 +79,7 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.when;
class CachingCatalogTest extends CatalogTestBase {
@@ -364,7 +365,7 @@ public void testCreatePartitionsWithIgnoreIfExistsInvalidatesPartitionCache() th
when(wrapped.listPartitions(identifier)).thenReturn(emptyList(), singletonList(created));
assertThat(catalog.listPartitions(identifier)).isEmpty();
- catalog.createPartitions(identifier, singletonList(spec), false, null, false);
+ catalog.createPartitions(identifier, singletonList(spec), false, null, false, null);
assertThat(catalog.listPartitions(identifier)).containsExactly(created);
}
@@ -383,15 +384,65 @@ public void testCreatePartitionsWithStatisticsForwardsAndInvalidatesPartitionCac
when(wrapped.listPartitions(identifier)).thenReturn(emptyList(), singletonList(created));
assertThat(catalog.listPartitions(identifier)).isEmpty();
- catalog.createPartitions(identifier, singletonList(spec), true, statistics, false);
+ catalog.createPartitions(identifier, singletonList(spec), true, statistics, false, null);
// Dropping the forward would leave the statistics unreported and nothing else would say so.
Mockito.verify(wrapped)
- .createPartitions(identifier, singletonList(spec), true, statistics, false);
+ .createPartitions(identifier, singletonList(spec), true, statistics, false, null);
// A report changes what a partition holds, so the cached listing is stale after it.
assertThat(catalog.listPartitions(identifier)).containsExactly(created);
}
+ @Test
+ public void testCreatePartitionsWithLocationForwardsAndInvalidatesPartitionCache()
+ throws Exception {
+ Catalog wrapped = Mockito.mock(Catalog.class);
+ TestableCachingCatalog catalog =
+ new TestableCachingCatalog(wrapped, EXPIRATION_TTL, ticker);
+ Identifier identifier = new Identifier("db", "tbl");
+ Map spec = singletonMap("dt", "20260717");
+ List locations = singletonList("file:/archive/dt=20260717");
+ Partition created = new Partition(spec, 0, 0, 0, 0, -1, false);
+ when(wrapped.listPartitions(identifier)).thenReturn(emptyList(), singletonList(created));
+
+ assertThat(catalog.listPartitions(identifier)).isEmpty();
+ catalog.createPartitions(identifier, singletonList(spec), true, null, false, locations);
+
+ Mockito.verify(wrapped)
+ .createPartitions(identifier, singletonList(spec), true, null, false, locations);
+ assertThat(catalog.listPartitions(identifier)).containsExactly(created);
+ }
+
+ @Test
+ public void testCreatePartitionsWithLocationInvalidatesCacheAfterConfirmationFailure()
+ throws Exception {
+ Catalog wrapped = Mockito.mock(Catalog.class);
+ TestableCachingCatalog catalog =
+ new TestableCachingCatalog(wrapped, EXPIRATION_TTL, ticker);
+ Identifier identifier = new Identifier("db", "tbl");
+ Map spec = singletonMap("dt", "20260717");
+ List locations = singletonList("file:/archive/dt=20260717");
+ Partition created = new Partition(spec, 0, 0, 0, 0, -1, false);
+ when(wrapped.listPartitions(identifier)).thenReturn(emptyList(), singletonList(created));
+ doThrow(new IllegalStateException("location confirmation failed"))
+ .when(wrapped)
+ .createPartitions(identifier, singletonList(spec), true, null, false, locations);
+
+ assertThat(catalog.listPartitions(identifier)).isEmpty();
+ assertThatThrownBy(
+ () ->
+ catalog.createPartitions(
+ identifier,
+ singletonList(spec),
+ true,
+ null,
+ false,
+ locations))
+ .hasMessage("location confirmation failed");
+
+ assertThat(catalog.listPartitions(identifier)).containsExactly(created);
+ }
+
@Test
public void testDeadlock() throws Exception {
Catalog underlyCatalog = this.catalog;
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java
index ec230c8504bb..051db58e5838 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/DelegateCatalogTest.java
@@ -53,9 +53,9 @@ void testCreatePartitionsCarriesStatisticsAndModeToTheWrappedCatalog() throws Ex
Collections.singletonList(
new PartitionStatistics(specs.get(0), 3L, 300L, 1L, 1000L, -1));
- delegating.createPartitions(IDENTIFIER, specs, true, statistics, false);
+ delegating.createPartitions(IDENTIFIER, specs, true, statistics, false, null);
- verify(wrapped).createPartitions(IDENTIFIER, specs, true, statistics, false);
+ verify(wrapped).createPartitions(IDENTIFIER, specs, true, statistics, false, null);
// Falling through to the two-argument call is how the statistics would go missing.
verify(wrapped, never()).createPartitions(any(), anyList());
}
@@ -67,9 +67,23 @@ void testCreatePartitionsCarriesTheAbsenceOfStatisticsThrough() throws Exception
List> specs =
Collections.singletonList(Collections.singletonMap("dt", "20260728"));
- delegating.createPartitions(IDENTIFIER, specs, false, null, false);
+ delegating.createPartitions(IDENTIFIER, specs, false, null, false, null);
- verify(wrapped).createPartitions(IDENTIFIER, specs, false, null, false);
+ verify(wrapped).createPartitions(IDENTIFIER, specs, false, null, false, null);
+ }
+
+ @Test
+ void testCreatePartitionsCarriesLocationsToTheWrappedCatalog() throws Exception {
+ Catalog wrapped = mock(Catalog.class);
+ Catalog delegating = new TestDelegateCatalog(wrapped);
+ Map spec = Collections.singletonMap("dt", "20260728");
+ List> specs = Collections.singletonList(spec);
+ List locations = Collections.singletonList("file:/archive/dt=20260728");
+
+ delegating.createPartitions(IDENTIFIER, specs, true, null, false, locations);
+
+ verify(wrapped).createPartitions(IDENTIFIER, specs, true, null, false, locations);
+ verify(wrapped, never()).createPartitions(IDENTIFIER, specs, true, null, false, null);
}
/** {@link DelegateCatalog} forwards every operation; these tests never rebuild one. */
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 7c5503534581..b452e05f2d3f 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
@@ -54,6 +54,7 @@
import org.apache.paimon.rest.exceptions.BadRequestException;
import org.apache.paimon.rest.exceptions.NotAuthorizedException;
import org.apache.paimon.rest.exceptions.NotImplementedException;
+import org.apache.paimon.rest.requests.CreatePartitionsRequest;
import org.apache.paimon.rest.responses.ConfigResponse;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaChange;
@@ -84,11 +85,18 @@
import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import static org.apache.paimon.catalog.Catalog.SYSTEM_DATABASE_NAME;
import static org.apache.paimon.catalog.Catalog.TABLE_DEFAULT_OPTION_PREFIX;
import static org.apache.paimon.rest.RESTApi.HEADER_PREFIX;
import static org.apache.paimon.rest.RESTApi.READ_VIA_HEADER;
+import static org.apache.paimon.utils.SnapshotManagerTest.createSnapshotWithMillis;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -486,6 +494,276 @@ void testCatalogManagedPartitionListingReflectsCatalogMutationsImmediately() thr
assertThat(partitionManager.listPartitions(Collections.emptyMap(), null)).isEmpty();
}
+ @Test
+ void testCustomPartitionLocationUsesExistingRouteAndStoresCanonicalLocation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ String requested = "OSS://ARCHIVE-BUCKET//history///%64t%3D20260717/";
+ String canonical = "oss://archive-bucket/history/dt=20260717";
+ String partitionsResource =
+ ResourcePaths.forCatalogProperties(restCatalog.api().options())
+ .partitions(identifier.getDatabaseName(), identifier.getObjectName());
+ restCatalogServer.clearReceivedHeaders();
+
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ Collections.singletonList(requested));
+
+ assertThat(restCatalogServer.getReceivedHeaders(partitionsResource)).hasSize(1);
+ assertThat(onlyPartition(identifier).location()).isEqualTo(canonical);
+ }
+
+ @Test
+ void testRenamePreservesCustomPartitionLocationAndSnapshotState() throws Exception {
+ Identifier source = createFormatTableWithCatalogManagedPartitions();
+ Identifier destination =
+ Identifier.create(source.getDatabaseName(), "renamed_managed_partition_table");
+ Map spec = Collections.singletonMap("dt", "20260717");
+ String location = "file:/archive/dt=20260717";
+ Snapshot snapshot = createSnapshotWithMillis(1L, System.currentTimeMillis());
+ restCatalog.createPartitions(
+ source,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ Collections.singletonList(location));
+ restCatalogServer.setTableSnapshot(source, snapshot, 1L, 2L, 3L, 4L);
+
+ restCatalog.renameTable(source, destination, false);
+
+ assertThat(restCatalog.listPartitions(destination))
+ .singleElement()
+ .satisfies(
+ partition -> {
+ assertThat(partition.spec()).isEqualTo(spec);
+ assertThat(partition.location()).isEqualTo(location);
+ });
+ assertThat(restCatalog.loadSnapshot(destination))
+ .get()
+ .satisfies(
+ tableSnapshot -> {
+ assertThat(tableSnapshot.snapshot().id()).isEqualTo(snapshot.id());
+ assertThat(tableSnapshot.recordCount()).isEqualTo(1L);
+ });
+ }
+
+ @Test
+ void testInvalidCustomPartitionLocationFailsBeforePost() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ String partitionsResource =
+ ResourcePaths.forCatalogProperties(restCatalog.api().options())
+ .partitions(identifier.getDatabaseName(), identifier.getObjectName());
+ restCatalogServer.clearReceivedHeaders();
+
+ assertThatThrownBy(
+ () ->
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ Collections.singletonList(
+ "oss://archive-bucket/history/%2e%2e/secret")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Invalid custom partition location");
+
+ assertThat(restCatalogServer.getReceivedHeaders(partitionsResource)).isEmpty();
+ }
+
+ @Test
+ void testAlignedCustomPartitionLocationsRejectInvalidRequestsBeforeMutation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map first = Collections.singletonMap("dt", "20260717");
+ Map second = Collections.singletonMap("dt", "20260718");
+ String resource =
+ ResourcePaths.forCatalogProperties(restCatalog.api().options())
+ .partitions(identifier.getDatabaseName(), identifier.getObjectName());
+ HttpClient client = new HttpClient(restCatalogServer.getUrl());
+
+ assertThatThrownBy(
+ () ->
+ client.post(
+ resource,
+ new MisalignedCreatePartitionsRequest(
+ Arrays.asList(first, second),
+ Collections.singletonList(
+ "file:/archive/dt=20260717")),
+ restCatalog.api().authFunction()))
+ .isInstanceOf(BadRequestException.class)
+ .hasMessageContaining("same size as partitionSpecs");
+ assertThatThrownBy(
+ () ->
+ client.post(
+ resource,
+ new CreatePartitionsRequest(
+ Collections.singletonList(first),
+ true,
+ null,
+ null,
+ Collections.singletonList(" ")),
+ restCatalog.api().authFunction()))
+ .isInstanceOf(BadRequestException.class)
+ .hasMessageContaining("non-blank");
+ assertThatThrownBy(
+ () ->
+ client.post(
+ resource,
+ new CreatePartitionsRequest(
+ Arrays.asList(first, first),
+ true,
+ null,
+ null,
+ Arrays.asList(
+ "file:/archive/one", "file:/archive/two")),
+ restCatalog.api().authFunction()))
+ .isInstanceOf(BadRequestException.class)
+ .hasMessageContaining("must not contain duplicates");
+ assertThat(restCatalog.listPartitions(identifier)).isEmpty();
+ }
+
+ @Test
+ void testCustomPartitionLocationOwnershipConflictsAreRejectedAtomically() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map first = Collections.singletonMap("dt", "20260717");
+ Map second = Collections.singletonMap("dt", "20260718");
+ List> conflicts =
+ Arrays.asList(
+ Arrays.asList("file:/archive/shared", "file:/archive/shared"),
+ Arrays.asList("file:/archive/root", "file:/archive/root/nested"),
+ Arrays.asList("file:/archive/root/nested", "file:/archive/root"));
+
+ for (List locations : conflicts) {
+ for (boolean ignoreIfExists : Arrays.asList(true, false)) {
+ assertThatThrownBy(
+ () ->
+ restCatalog.createPartitions(
+ identifier,
+ Arrays.asList(first, second),
+ ignoreIfExists,
+ null,
+ false,
+ locations))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThat(restCatalog.listPartitions(identifier)).isEmpty();
+ }
+ }
+ }
+
+ @Test
+ void testConcurrentCustomLocationCreatesValidateAndCommitSerially() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map firstSpec = Collections.singletonMap("dt", "20260717");
+ Map secondSpec = Collections.singletonMap("dt", "20260718");
+ String sharedLocation = "file:/archive/shared";
+ CountDownLatch start = new CountDownLatch(1);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+
+ try {
+ Future first =
+ executor.submit(
+ () -> {
+ start.await();
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(firstSpec),
+ true,
+ null,
+ false,
+ Collections.singletonList(sharedLocation));
+ return null;
+ });
+ Future second =
+ executor.submit(
+ () -> {
+ start.await();
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(secondSpec),
+ true,
+ null,
+ false,
+ Collections.singletonList(sharedLocation));
+ return null;
+ });
+
+ start.countDown();
+ int failures = 0;
+ for (Future future : Arrays.asList(first, second)) {
+ try {
+ future.get(10, TimeUnit.SECONDS);
+ } catch (ExecutionException e) {
+ assertThat(e).hasCauseInstanceOf(IllegalArgumentException.class);
+ failures++;
+ }
+ }
+ assertThat(failures).isEqualTo(1);
+ List stored = restCatalog.listPartitions(identifier);
+ assertThat(stored).hasSize(1);
+ assertThat(stored.get(0).location()).isEqualTo(sharedLocation);
+ } finally {
+ start.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void testUnsupportedCustomPartitionLocationCreateFailsWithoutMutation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ String resource =
+ ResourcePaths.forCatalogProperties(restCatalog.api().options())
+ .partitions(identifier.getDatabaseName(), identifier.getObjectName());
+ restCatalogServer.setPartitionLocationCreateSupported(false);
+ restCatalogServer.clearReceivedHeaders();
+
+ assertThatThrownBy(
+ () ->
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ Collections.singletonList("file:/archive/dt=20260717")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("does not support custom partition locations");
+
+ assertThat(restCatalogServer.getReceivedHeaders(resource)).hasSize(1);
+ assertThat(restCatalog.listPartitions(identifier)).isEmpty();
+ }
+
+ @Test
+ void testServerCanonicalizesCustomAndDerivedLocations() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map first = Collections.singletonMap("dt", "20260717");
+ Map second = Collections.singletonMap("dt", "20260718");
+ String resource =
+ ResourcePaths.forCatalogProperties(restCatalog.api().options())
+ .partitions(identifier.getDatabaseName(), identifier.getObjectName());
+ HttpClient client = new HttpClient(restCatalogServer.getUrl());
+
+ client.post(
+ resource,
+ new CreatePartitionsRequest(
+ Arrays.asList(first, second),
+ true,
+ null,
+ null,
+ Arrays.asList("FILE:///archive//%64t%3D20260717/", null)),
+ restCatalog.api().authFunction());
+
+ assertThat(restCatalog.listPartitions(identifier))
+ .extracting(Partition::location)
+ .containsExactlyInAnyOrder("file:/archive/dt=20260717", null);
+ }
+
@Test
void testPartitionManagerSurvivesSerialization() throws Exception {
Identifier identifier = createFormatTableWithCatalogManagedPartitions();
@@ -526,7 +804,8 @@ void testReportedPartitionStatisticsAreStoredAndReadBack() throws Exception {
true,
Collections.singletonList(
new PartitionStatistics(spec, 3L, 300L, 1L, 1000L, -1)),
- false);
+ false,
+ null);
assertStatistics(identifier, 3L, 300L, 1L, 1000L);
// ADD again, through the partition manager a writer commits with: the counts accumulate
@@ -535,7 +814,8 @@ void testReportedPartitionStatisticsAreStoredAndReadBack() throws Exception {
specs,
true,
Collections.singletonList(new PartitionStatistics(spec, 4L, 400L, 2L, 500L, -1)),
- false);
+ false,
+ null);
assertStatistics(identifier, 7L, 700L, 3L, 1000L);
// A field reported as unknown leaves the stored one alone rather than zeroing it.
@@ -551,7 +831,8 @@ void testReportedPartitionStatisticsAreStoredAndReadBack() throws Exception {
PartitionStatistics.UNKNOWN,
PartitionStatistics.UNKNOWN,
-1)),
- false);
+ false,
+ null);
assertStatistics(identifier, 7L, 800L, 3L, 1000L);
// SET is the whole partition now: every reported field is replaced, including a creation
@@ -564,7 +845,8 @@ void testReportedPartitionStatisticsAreStoredAndReadBack() throws Exception {
true,
Collections.singletonList(
new PartitionStatistics(spec, 5L, 500L, 1L, 700L, -1)),
- true);
+ true,
+ null);
assertStatistics(identifier, 5L, 500L, 1L, 700L);
// Unknown is skipped under SET too: it reports nothing about that field, not a zero.
@@ -580,7 +862,8 @@ void testReportedPartitionStatisticsAreStoredAndReadBack() throws Exception {
PartitionStatistics.UNKNOWN,
PartitionStatistics.UNKNOWN,
-1)),
- true);
+ true,
+ null);
assertStatistics(identifier, 5L, 900L, 1L, 700L);
// Reporting never registers or unregisters anything.
@@ -606,7 +889,8 @@ void testStatisticsOfAnUnstoredPartitionAreDropped() throws Exception {
3L,
1000L,
-1)),
- false);
+ false,
+ null);
assertThat(restCatalog.listPartitions(identifier))
.extracting(Partition::spec)
@@ -628,7 +912,8 @@ void testAReportThatOnlyPartlyMatchesIsNotAppliedAtAll() throws Exception {
Arrays.asList(
new PartitionStatistics(stored, 3L, 300L, 1L, 1000L, -1),
new PartitionStatistics(absent, 9L, 900L, 3L, 2000L, -1)),
- false);
+ false,
+ null);
// Applying the half that matched would count it twice on the next report.
Partition partition = onlyPartition(identifier);
@@ -741,22 +1026,28 @@ void testRoundTrippedFormatTableReplacePassesClientValidation() throws Exception
}
private Identifier createFormatTableWithCatalogManagedPartitions() throws Exception {
+ return createFormatTableWithCatalogManagedPartitions(restCatalog);
+ }
+
+ private Identifier createFormatTableWithCatalogManagedPartitions(RESTCatalog catalog)
+ throws Exception {
Identifier identifier = Identifier.create("db1", "managed_partition_table");
- restCatalog.createDatabase(identifier.getDatabaseName(), true);
- restCatalog.createTable(
- identifier,
- Schema.newBuilder()
- .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString())
- .option(CoreOptions.METASTORE_PARTITIONED_TABLE.key(), "true")
- .option(CoreOptions.FILE_FORMAT.key(), "parquet")
- .column("id", DataTypes.INT())
- .column("dt", DataTypes.STRING())
- .partitionKeys("dt")
- .build(),
- false);
+ catalog.createDatabase(identifier.getDatabaseName(), true);
+ catalog.createTable(identifier, catalogManagedFormatTableSchema(), false);
return identifier;
}
+ private static Schema catalogManagedFormatTableSchema() {
+ return Schema.newBuilder()
+ .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString())
+ .option(CoreOptions.METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .build();
+ }
+
private static Predicate partitionFilter(String value) {
return new PredicateBuilder(
RowType.of(
@@ -1032,7 +1323,7 @@ private RESTCatalog initCatalogUtil(
defaultConf.put(
TABLE_DEFAULT_OPTION_PREFIX + createTableDefaultKey, createTableDefaultValue);
}
- this.config = new ConfigResponse(defaultConf, ImmutableMap.of());
+ this.config = new ConfigResponse(defaultConf, new HashMap<>());
restCatalogServer =
new RESTCatalogServer(dataPath, this.authProvider, this.config, restWarehouse);
restCatalogServer.start();
@@ -1052,6 +1343,33 @@ private RESTCatalog initCatalogUtil(
return new RESTCatalog(CatalogContext.create(options));
}
+ private static class MisalignedCreatePartitionsRequest implements RESTRequest {
+
+ private final List> partitionSpecs;
+ private final List partitionLocations;
+
+ private MisalignedCreatePartitionsRequest(
+ List> partitionSpecs, List partitionLocations) {
+ this.partitionSpecs = partitionSpecs;
+ this.partitionLocations = partitionLocations;
+ }
+
+ @JsonGetter("partitionSpecs")
+ public List> getPartitionSpecs() {
+ return partitionSpecs;
+ }
+
+ @JsonGetter("partitionLocations")
+ public List getPartitionLocations() {
+ return partitionLocations;
+ }
+
+ @JsonGetter("ignoreIfExists")
+ public boolean ignoreIfExists() {
+ return true;
+ }
+ }
+
private static class InvalidColumnGrantRequest implements RESTRequest {
private final PermissionResource resource;
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 44449fcabcec..ac05558e1387 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
@@ -59,6 +59,7 @@
import org.junit.Test;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -66,6 +67,7 @@
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/** Test for {@link RESTApi} json. */
@@ -337,6 +339,34 @@ public void createPartitionsRequestParseTest() throws Exception {
assertNull(defaultRequest.replaceStatistics());
}
+ @Test
+ public void createPartitionsRequestPreservesLocationsTest() throws Exception {
+ String json =
+ "{\"partitionSpecs\":[{\"dt\":\"20260901\"},{\"dt\":\"20260902\"}],"
+ + "\"partitionLocations\":[null,"
+ + "\"oss://archive-bucket/table/dt=20260902\"]}";
+
+ CreatePartitionsRequest request = RESTApi.fromJson(json, CreatePartitionsRequest.class);
+
+ assertEquals(
+ Arrays.asList(null, "oss://archive-bucket/table/dt=20260902"),
+ request.getPartitionLocations());
+ assertEquals(
+ request.getPartitionLocations(),
+ RESTApi.fromJson(RESTApi.toJson(request), CreatePartitionsRequest.class)
+ .getPartitionLocations());
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new CreatePartitionsRequest(
+ request.getPartitionSpecs(),
+ true,
+ null,
+ null,
+ Collections.singletonList(null)));
+ }
+
@Test
public void createPartitionsRequestCarriesStatisticsTest() throws Exception {
Map spec = Collections.singletonMap("dt", "20260728");
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
new file mode 100644
index 000000000000..56da07538982
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java
@@ -0,0 +1,255 @@
+/*
+ * 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.rest;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.TableMetadata;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.rest.requests.CreatePartitionsRequest;
+import org.apache.paimon.table.format.FormatTablePartitionPathResolver;
+import org.apache.paimon.table.format.FormatTablePartitionRegistryValidator;
+import org.apache.paimon.utils.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.CoreOptions.PATH;
+
+/** Helpers for validating and updating partition state in the mock REST catalog. */
+final class RESTCatalogPartitionSupport {
+
+ private RESTCatalogPartitionSupport() {}
+
+ @Nullable
+ static List canonicalizeRequestedLocations(
+ CreatePartitionsRequest request, CatalogContext catalogContext) {
+ List locations = request.getPartitionLocations();
+ if (locations == null) {
+ return null;
+ }
+ List> specs = request.getPartitionSpecs();
+ if (specs == null || locations.size() != specs.size()) {
+ throw new IllegalArgumentException(
+ "partitionLocations must contain exactly one entry for every partition spec.");
+ }
+ Set> uniqueSpecs = new HashSet<>();
+ List canonical = new ArrayList<>(locations.size());
+ for (int i = 0; i < locations.size(); i++) {
+ if (specs.get(i) == null || !uniqueSpecs.add(specs.get(i))) {
+ throw new IllegalArgumentException(
+ "partitionSpecs must not contain duplicates when partitionLocations is present.");
+ }
+ String location = locations.get(i);
+ if (location == null) {
+ canonical.add(null);
+ continue;
+ }
+ if (StringUtils.isBlank(location)) {
+ throw new IllegalArgumentException(
+ "partitionLocations must contain null or a non-blank absolute location.");
+ }
+ canonical.add(
+ FormatTablePartitionPathResolver.canonicalizeCustomLocation(
+ location, catalogContext)
+ .toString());
+ }
+ return canonical;
+ }
+
+ static void validateFormatTablePartitionLocations(
+ List partitions,
+ TableMetadata metadata,
+ String tableName,
+ CatalogContext catalogContext) {
+ if (partitions.stream().noneMatch(partition -> partition.location() != null)) {
+ return;
+ }
+ 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));
+ }
+ try {
+ FormatTablePartitionRegistryValidator.validatePartitionLocations(
+ partitions,
+ metadata.schema().partitionKeys(),
+ new Path(tablePath),
+ tableName,
+ new CoreOptions(metadata.schema().options())
+ .formatTablePartitionOnlyValueInPath(),
+ catalogContext);
+ } catch (IllegalStateException e) {
+ throw new IllegalArgumentException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Folds reported statistics into stored partitions. Applying is all-or-nothing when a report
+ * names an unknown partition, because applying only part would double-count on a retry.
+ */
+ static void applyStatistics(
+ List storedPartitions,
+ @Nullable List statistics,
+ @Nullable Boolean replaceStatistics) {
+ if (statistics == null) {
+ return;
+ }
+ boolean accumulate = !Boolean.TRUE.equals(replaceStatistics);
+ Map, PartitionStatistics> reported = new HashMap<>();
+ for (PartitionStatistics statistic : statistics) {
+ reported.put(statistic.spec(), statistic);
+ }
+ Set> storedSpecs =
+ storedPartitions.stream().map(Partition::spec).collect(Collectors.toSet());
+ if (!storedSpecs.containsAll(reported.keySet())) {
+ return;
+ }
+ for (int i = 0; i < storedPartitions.size(); i++) {
+ Partition stored = storedPartitions.get(i);
+ PartitionStatistics update = reported.get(stored.spec());
+ if (update == null) {
+ continue;
+ }
+ storedPartitions.set(i, mergeStatistics(stored, update, accumulate));
+ }
+ }
+
+ /** Returns the post-commit partition snapshot without mutating the stored list. */
+ @Nullable
+ static List mergeSnapshotStatistics(
+ @Nullable List storedPartitions,
+ @Nullable List statistics) {
+ if (storedPartitions == null && statistics == null) {
+ return null;
+ }
+
+ List merged =
+ storedPartitions == null ? new ArrayList<>() : new ArrayList<>(storedPartitions);
+ if (statistics != null) {
+ Map, PartitionStatistics> reported = new HashMap<>();
+ for (PartitionStatistics statistic : statistics) {
+ reported.putIfAbsent(statistic.spec(), statistic);
+ }
+ Set> existingSpecs = new HashSet<>();
+ for (int i = 0; i < merged.size(); i++) {
+ Partition stored = merged.get(i);
+ existingSpecs.add(stored.spec());
+ PartitionStatistics update = reported.get(stored.spec());
+ if (update != null) {
+ merged.set(i, mergeSnapshotStatistics(stored, update));
+ }
+ }
+ for (PartitionStatistics update : statistics) {
+ if (!existingSpecs.contains(update.spec())) {
+ merged.add(newPartition(update));
+ }
+ }
+ }
+ merged.removeIf(
+ partition ->
+ partition.fileSizeInBytes() <= 0
+ && partition.fileCount() <= 0
+ && partition.recordCount() <= 0);
+ return merged;
+ }
+
+ private static Partition mergeStatistics(
+ Partition stored, PartitionStatistics update, boolean accumulate) {
+ return new Partition(
+ stored.spec(),
+ combine(stored.recordCount(), update.recordCount(), accumulate),
+ combine(stored.fileSizeInBytes(), update.fileSizeInBytes(), accumulate),
+ combine(stored.fileCount(), update.fileCount(), accumulate),
+ combineLastFileCreationTime(
+ stored.lastFileCreationTime(), update.lastFileCreationTime(), accumulate),
+ stored.totalBuckets(),
+ stored.done(),
+ stored.createdAt(),
+ stored.createdBy(),
+ stored.updatedAt(),
+ stored.updatedBy(),
+ stored.options(),
+ stored.location());
+ }
+
+ private static Partition mergeSnapshotStatistics(Partition stored, PartitionStatistics update) {
+ return new Partition(
+ stored.spec(),
+ accumulateDelta(stored.recordCount(), update.recordCount()),
+ accumulateDelta(stored.fileSizeInBytes(), update.fileSizeInBytes()),
+ accumulateDelta(stored.fileCount(), update.fileCount()),
+ Math.max(stored.lastFileCreationTime(), update.lastFileCreationTime()),
+ update.totalBuckets(),
+ stored.done(),
+ stored.createdAt(),
+ stored.createdBy(),
+ stored.updatedAt(),
+ stored.updatedBy(),
+ stored.options(),
+ stored.location());
+ }
+
+ private static Partition newPartition(PartitionStatistics statistics) {
+ return new Partition(
+ statistics.spec(),
+ statistics.recordCount(),
+ statistics.fileSizeInBytes(),
+ statistics.fileCount(),
+ statistics.lastFileCreationTime(),
+ statistics.totalBuckets(),
+ false,
+ System.currentTimeMillis(),
+ "created",
+ System.currentTimeMillis(),
+ "updated",
+ new HashMap<>());
+ }
+
+ private static long accumulateDelta(long stored, long reported) {
+ return PartitionStatistics.isKnown(stored) ? stored + reported : reported;
+ }
+
+ private static long combine(long stored, long reported, boolean accumulate) {
+ if (!PartitionStatistics.isKnown(reported)) {
+ return stored;
+ }
+ if (!accumulate || !PartitionStatistics.isKnown(stored)) {
+ return reported;
+ }
+ return stored + reported;
+ }
+
+ private static long combineLastFileCreationTime(
+ long stored, long reported, boolean accumulate) {
+ if (!PartitionStatistics.isKnown(reported)) {
+ return stored;
+ }
+ return accumulate ? Math.max(stored, reported) : reported;
+ }
+}
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 8e36aa384401..ab53d38b0922 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
@@ -177,6 +177,7 @@
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -252,10 +253,11 @@ public class RESTCatalogServer {
private final Queue scriptedListPartitionsByFilterResponses =
new ConcurrentLinkedQueue<>();
- private final Map> tablePartitionsStore = new HashMap<>();
+ private final Map> tablePartitionsStore = new ConcurrentHashMap<>();
private final Map viewStore = new ConcurrentHashMap<>();
- private final Map tableLatestSnapshotStore = new HashMap<>();
- private final Map tableWithSnapshotId2SnapshotStore = new HashMap<>();
+ private final Map tableLatestSnapshotStore = new ConcurrentHashMap<>();
+ private final Map, TableSnapshot> tableWithSnapshotId2SnapshotStore =
+ new ConcurrentHashMap<>();
private final List noPermissionDatabases = new ArrayList<>();
private final List noPermissionTables = new ArrayList<>();
private final List noPermissionViews = new ArrayList<>();
@@ -268,10 +270,12 @@ public class RESTCatalogServer {
private final ResourcePaths resourcePaths;
- private final List> receivedHeaders = new ArrayList<>();
- private final Map>> receivedHeadersByPath = new HashMap<>();
+ private final List> receivedHeaders = new CopyOnWriteArrayList<>();
+ private final Map>> receivedHeadersByPath =
+ new ConcurrentHashMap<>();
private volatile boolean partitionListingSupported = true;
+ private volatile boolean partitionLocationCreateSupported = true;
public RESTCatalogServer(
String dataPath, AuthProvider authProvider, ConfigResponse config, String warehouse) {
@@ -325,7 +329,7 @@ public void setTableSnapshot(
snapshot, recordCount, fileSizeInBytes, fileCount, lastFileCreationTime);
tableLatestSnapshotStore.put(identifier.getFullName(), tableSnapshot);
tableWithSnapshotId2SnapshotStore.put(
- geTableFullNameWithSnapshotId(identifier, snapshot.id()), tableSnapshot);
+ tableSnapshotKey(identifier, snapshot.id()), tableSnapshot);
}
public void setDataToken(Identifier identifier, RESTToken token) {
@@ -340,6 +344,10 @@ public void setPartitionListingSupported(boolean partitionListingSupported) {
this.partitionListingSupported = partitionListingSupported;
}
+ public void setPartitionLocationCreateSupported(boolean partitionLocationCreateSupported) {
+ this.partitionLocationCreateSupported = partitionLocationCreateSupported;
+ }
+
public void clearReceivedListPartitionsByFilterRequests() {
receivedListPartitionsByFilterRequests.clear();
}
@@ -433,7 +441,7 @@ public MockResponse dispatch(RecordedRequest request) {
String[] paths = request.getPath().split("\\?");
String resourcePath = paths[0];
receivedHeadersByPath
- .computeIfAbsent(resourcePath, ignored -> new ArrayList<>())
+ .computeIfAbsent(resourcePath, ignored -> new CopyOnWriteArrayList<>())
.add(new HashMap<>(headers));
Map parameters =
paths.length == 2 ? getParameters(paths[1]) : Collections.emptyMap();
@@ -478,11 +486,9 @@ && isTableByIdRequest(request.getPath())) {
} else if (StringUtils.startsWith(
request.getPath(), resourcePaths.functions())) {
return functionsHandle(parameters);
- } else if (request.getPath().startsWith(databaseUri)) {
+ } else if (resourcePath.startsWith(databaseUri)) {
String[] resources =
- request.getPath()
- .substring((databaseUri + "/").length())
- .split("/");
+ resourcePath.substring((databaseUri + "/").length()).split("/");
String databaseName = RESTUtil.decodeString(resources[0]);
if (noPermissionDatabases.contains(databaseName)) {
throw new Catalog.DatabaseNoPermissionException(databaseName);
@@ -561,7 +567,7 @@ && isTableByIdRequest(request.getPath())) {
boolean isPartitions =
resources.length == 4
&& ResourcePaths.TABLES.equals(resources[1])
- && resources[3].startsWith("partitions");
+ && "partitions".equals(resources[3]);
boolean isMarkDonePartitions =
resources.length == 5
@@ -584,6 +590,12 @@ && isTableByIdRequest(request.getPath())) {
&& ResourcePaths.TABLES.equals(resources[1])
&& "partitions".equals(resources[3])
&& "drop".equals(resources[4]);
+ boolean isPartitionOperation =
+ isPartitions
+ || isMarkDonePartitions
+ || isDropPartitions
+ || isListPartitionsByNames
+ || isListPartitionsByFilter;
boolean isBranches =
resources.length >= 4
@@ -613,47 +625,49 @@ && isTableByIdRequest(request.getPath())) {
throw new Catalog.TableNoPermissionException(identifier);
}
}
- // validate partition
- if (isPartitions || isMarkDonePartitions || isDropPartitions) {
- String tableName = RESTUtil.decodeString(resources[2]);
- Optional error =
- checkTablePartitioned(
- Identifier.create(databaseName, tableName));
- if (error.isPresent()) {
- return error.get();
+ if (isPartitionOperation) {
+ synchronized (tableLifecycleLocks.lock(identifier.getFullName())) {
+ Optional error = checkTablePartitioned(identifier);
+ if (error.isPresent()) {
+ return error.get();
+ }
+ if (!partitionListingSupported
+ && ((isPartitions
+ && "GET".equals(restAuthParameter.method()))
+ || isListPartitionsByNames
+ || isListPartitionsByFilter)) {
+ return mockResponse(
+ new ErrorResponse(null, null, "", 501), 501);
+ }
+ if (isMarkDonePartitions) {
+ MarkDonePartitionsRequest requestBody =
+ parseRequest(data, MarkDonePartitionsRequest.class);
+ catalog.markDonePartitions(
+ identifier, requestBody.getPartitionSpecs());
+ return new MockResponse().setResponseCode(200);
+ }
+ if (isDropPartitions) {
+ return dropPartitionsHandle(
+ restAuthParameter.data(), identifier);
+ }
+ if (isPartitions) {
+ return partitionsApiHandle(
+ restAuthParameter.method(),
+ restAuthParameter.data(),
+ parameters,
+ identifier);
+ }
+ if (isListPartitionsByFilter) {
+ return listPartitionsByFilter(
+ identifier,
+ parseRequest(
+ data, ListPartitionsByFilterRequest.class));
+ }
+ ListPartitionsByNamesRequest requestBody =
+ parseRequest(data, ListPartitionsByNamesRequest.class);
+ return listPartitionsByNames(
+ parameters, identifier, requestBody.getPartitionSpecs());
}
- }
- if (isMarkDonePartitions) {
- MarkDonePartitionsRequest markDonePartitionsRequest =
- parseRequest(data, MarkDonePartitionsRequest.class);
- catalog.markDonePartitions(
- identifier, markDonePartitionsRequest.getPartitionSpecs());
- return new MockResponse().setResponseCode(200);
- } else if (!partitionListingSupported
- && ((isPartitions && "GET".equals(restAuthParameter.method()))
- || isListPartitionsByNames
- || isListPartitionsByFilter)) {
- return mockResponse(new ErrorResponse(null, null, "", 501), 501);
- } else if (isDropPartitions) {
- return dropPartitionsHandle(restAuthParameter.data(), identifier);
- } else if (isPartitions) {
- return partitionsApiHandle(
- restAuthParameter.method(),
- restAuthParameter.data(),
- parameters,
- identifier);
- } else if (isListPartitionsByFilter) {
- ListPartitionsByFilterRequest listPartitionsByFilterRequest =
- parseRequest(data, ListPartitionsByFilterRequest.class);
- return listPartitionsByFilter(
- identifier, listPartitionsByFilterRequest);
- } else if (isListPartitionsByNames) {
- ListPartitionsByNamesRequest listPartitionsByNamesRequest =
- parseRequest(data, ListPartitionsByNamesRequest.class);
- return listPartitionsByNames(
- parameters,
- identifier,
- listPartitionsByNamesRequest.getPartitionSpecs());
} else if (isBranches) {
return branchApiHandle(
resources,
@@ -682,37 +696,43 @@ && isTableByIdRequest(request.getPath())) {
} else if (isTableAuth) {
return authTable(identifier, restAuthParameter.data());
} else if (isCommitSnapshot) {
- return commitTableHandle(identifier, restAuthParameter.data());
+ synchronized (tableLifecycleLocks.lock(identifier.getFullName())) {
+ return commitTableHandle(identifier, restAuthParameter.data());
+ }
} else if (isRollbackTable) {
RollbackTableRequest requestBody =
parseRequest(data, RollbackTableRequest.class);
- if (noPermissionTables.contains(identifier.getFullName())) {
- throw new Catalog.TableNoPermissionException(identifier);
- }
- if (!tableMetadataStore.containsKey(identifier.getFullName())) {
- throw new Catalog.TableNotExistException(identifier);
- }
- if (requestBody.getInstant() instanceof Instant.SnapshotInstant) {
- long snapshotId =
- ((Instant.SnapshotInstant) requestBody.getInstant())
- .getSnapshotId();
- return rollbackTableByIdHandle(
- identifier, snapshotId, requestBody.getFromSnapshot());
- } else if (requestBody.getInstant() instanceof Instant.TagInstant) {
+ synchronized (tableLifecycleLocks.lock(identifier.getFullName())) {
+ if (noPermissionTables.contains(identifier.getFullName())) {
+ throw new Catalog.TableNoPermissionException(identifier);
+ }
+ if (!tableMetadataStore.containsKey(identifier.getFullName())) {
+ throw new Catalog.TableNotExistException(identifier);
+ }
+ if (requestBody.getInstant() instanceof Instant.SnapshotInstant) {
+ long snapshotId =
+ ((Instant.SnapshotInstant) requestBody.getInstant())
+ .getSnapshotId();
+ return rollbackTableByIdHandle(
+ identifier, snapshotId, requestBody.getFromSnapshot());
+ }
String tagName =
((Instant.TagInstant) requestBody.getInstant())
.getTagName();
return rollbackTableByTagNameHandle(identifier, tagName);
}
} else if (isRollbackSchema) {
- return rollbackSchemaHandle(identifier, restAuthParameter.data());
+ synchronized (tableLifecycleLocks.lock(identifier.getFullName())) {
+ return rollbackSchemaHandle(identifier, restAuthParameter.data());
+ }
} else if (isReplaceTable) {
return replaceTableHandle(identifier, restAuthParameter.data());
} else if (isTable) {
return tableHandle(
restAuthParameter.method(),
restAuthParameter.data(),
- identifier);
+ identifier,
+ null);
} else if (isTables) {
return tablesHandle(
restAuthParameter.method(),
@@ -1047,19 +1067,16 @@ private MockResponse loadSnapshot(Identifier identifier, String version) throws
}
private Optional checkTablePartitioned(Identifier identifier) {
- if (tableMetadataStore.containsKey(identifier.getFullName())) {
- TableMetadata tableMetadata = tableMetadataStore.get(identifier.getFullName());
- boolean partitioned =
- CoreOptions.fromMap(tableMetadata.schema().options())
- .partitionedTableInMetastore();
- if (!partitioned) {
- return Optional.of(mockResponse(new ErrorResponse(null, null, "", 501), 501));
- }
- return Optional.empty();
+ TableMetadata tableMetadata = tableMetadataStore.get(identifier.getFullName());
+ if (tableMetadata == null) {
+ return Optional.of(
+ mockResponse(
+ new ErrorResponse(ErrorResponse.RESOURCE_TYPE_TABLE, null, "", 404),
+ 404));
}
- return Optional.of(
- mockResponse(
- new ErrorResponse(ErrorResponse.RESOURCE_TYPE_TABLE, null, "", 404), 404));
+ return CoreOptions.fromMap(tableMetadata.schema().options()).partitionedTableInMetastore()
+ ? Optional.empty()
+ : Optional.of(mockResponse(new ErrorResponse(null, null, "", 501), 501));
}
private MockResponse authTable(Identifier identifier, String data) throws Exception {
@@ -1190,7 +1207,7 @@ private MockResponse commitTableHandle(Identifier identifier, String data) throw
private MockResponse rollbackTableByIdHandle(
Identifier identifier, long snapshotId, @Nullable Long fromSnapshot) throws Exception {
FileStoreTable table = getFileTable(identifier);
- String identifierWithSnapshotId = geTableFullNameWithSnapshotId(identifier, snapshotId);
+ Pair identifierWithSnapshotId = tableSnapshotKey(identifier, snapshotId);
TableSnapshot toSnapshot = tableWithSnapshotId2SnapshotStore.get(identifierWithSnapshotId);
if (toSnapshot == null) {
return mockResponse(
@@ -1224,8 +1241,8 @@ private MockResponse rollbackTableByTagNameHandle(Identifier identifier, String
boolean isExist = table.tagManager().tagExists(tagName);
if (isExist) {
Snapshot snapshot = table.tagManager().getOrThrow(tagName).trimToSnapshot();
- String identifierWithSnapshotId =
- geTableFullNameWithSnapshotId(identifier, snapshot.id());
+ Pair identifierWithSnapshotId =
+ tableSnapshotKey(identifier, snapshot.id());
if (tableWithSnapshotId2SnapshotStore.containsKey(identifierWithSnapshotId)) {
table =
table.copy(
@@ -1272,8 +1289,7 @@ private void cleanSnapshot(Identifier identifier, Long snapshotId, Long latestSn
throws IOException {
if (latestSnapshotId > snapshotId) {
for (long i = snapshotId + 1; i < latestSnapshotId + 1; i++) {
- tableWithSnapshotId2SnapshotStore.remove(
- geTableFullNameWithSnapshotId(identifier, i));
+ tableWithSnapshotId2SnapshotStore.remove(tableSnapshotKey(identifier, i));
}
}
}
@@ -1720,7 +1736,7 @@ private void removeDatabaseTableState(String databaseName) {
synchronized (policyLock(metadata.uuid())) {
if (tableMetadataStore.remove(tableName, metadata)) {
removePolicies(metadata.uuid());
- tableLatestSnapshotStore.remove(tableName);
+ removeSnapshotState(tableName);
tablePartitionsStore.remove(tableName);
}
}
@@ -1803,30 +1819,24 @@ private List listTables(String databaseName, Map