> 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 options.
*/
default void createPartitions(
Identifier identifier,
List> partitions,
boolean ignoreIfExists,
@Nullable List statistics,
- boolean replaceStatistics)
+ boolean replaceStatistics,
+ @Nullable List> partitionOptions)
throws TableNotExistException {
+ if (partitionOptions != null) {
+ if (partitionOptions.size() != partitions.size() || partitionOptions.contains(null)) {
+ throw new IllegalArgumentException(
+ "Partition options must contain one non-null map per partition.");
+ }
+ if (partitionOptions.stream().anyMatch(options -> !options.isEmpty())) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Catalog %s does not support partition options.",
+ 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..2f20d38fdea9 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> partitionOptions)
throws TableNotExistException {
wrapped.createPartitions(
- identifier, partitions, ignoreIfExists, statistics, replaceStatistics);
+ identifier,
+ partitions,
+ ignoreIfExists,
+ statistics,
+ replaceStatistics,
+ partitionOptions);
}
@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..ead9b3dd473c 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;
@@ -84,6 +85,7 @@
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -760,7 +762,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 +771,19 @@ public void createPartitions(
List> partitions,
boolean ignoreIfExists,
@Nullable List statistics,
- boolean replaceStatistics)
+ boolean replaceStatistics,
+ @Nullable List> partitionOptions)
throws TableNotExistException {
+ List> canonicalOptions =
+ canonicalizePartitionOptions(identifier, partitions, partitionOptions);
try {
api.createPartitions(
- identifier, partitions, ignoreIfExists, statistics, replaceStatistics);
+ identifier,
+ partitions,
+ ignoreIfExists,
+ statistics,
+ replaceStatistics,
+ canonicalOptions);
} catch (NoSuchResourceException e) {
throw new TableNotExistException(identifier);
} catch (ForbiddenException e) {
@@ -786,7 +796,78 @@ public void createPartitions(
identifier, e.getMessage()));
} catch (BadRequestException e) {
throw new IllegalArgumentException(e.getMessage());
+ } catch (NotImplementedException e) {
+ if (canonicalOptions == null) {
+ throw e;
+ }
+ throw new UnsupportedOperationException(
+ String.format(
+ "REST Catalog server does not support partition options for table %s.",
+ identifier.getFullName()),
+ e);
+ }
+ }
+
+ @Nullable
+ private List> canonicalizePartitionOptions(
+ Identifier identifier,
+ List> partitions,
+ @Nullable List> requested) {
+ if (requested == null) {
+ return null;
+ }
+ if (requested.size() != partitions.size()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition options for table %s must align with all %d partition specs, but found %d.",
+ identifier.getFullName(), partitions.size(), requested.size()));
+ }
+ Set> uniquePartitions = new HashSet<>();
+ List> canonical = new ArrayList<>(requested.size());
+ boolean hasOptions = false;
+ for (int i = 0; i < requested.size(); i++) {
+ Map partition = partitions.get(i);
+ if (partition == null || !uniquePartitions.add(partition)) {
+ throw new IllegalArgumentException(
+ "Partition specs must be non-null and unique when partition options are provided.");
+ }
+ Map options = requested.get(i);
+ if (options == null) {
+ throw new IllegalArgumentException("Partition options must not contain null maps.");
+ }
+ if (options.entrySet().stream()
+ .anyMatch(entry -> entry.getKey() == null || entry.getValue() == null)) {
+ throw new IllegalArgumentException(
+ "Partition options must not contain null keys or values.");
+ }
+ Map copied = new HashMap<>(options);
+ String location = copied.get(PATH.key());
+ if (location != null) {
+ try {
+ copied.put(
+ PATH.key(),
+ FormatTablePartitionPathResolver.canonicalizeCustomLocation(
+ location, context)
+ .toString());
+ } catch (IllegalArgumentException e) {
+ throw invalidPartitionLocation(identifier, partition, e);
+ }
+ }
+ hasOptions |= !copied.isEmpty();
+ canonical.add(copied);
}
+ return hasOptions ? canonical : null;
+ }
+
+ 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..449ef83ab597 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,8 +159,9 @@ public void createPartitions(
List> partitions,
boolean ignoreIfExists,
@Nullable List statistics,
- boolean replaceStatistics) {
- // Validated before the empty check: returning early would swallow a malformed report.
+ boolean replaceStatistics,
+ @Nullable List> partitionOptions) {
+ validatePartitionOptions(partitionOptions, partitions);
Map, PartitionStatistics> statisticsBySpec =
validateAndIndexStatistics(statistics, partitions);
if (partitions.isEmpty()) {
@@ -172,25 +173,61 @@ 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,
+ partitionOptions);
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.
+ int offset = 0;
for (List> batch : batches(partitions)) {
// A partition and its statistics travel in the same request.
+ int end = offset + batch.size();
catalog.createPartitions(
identifier,
batch,
true,
statisticsOf(batch, statisticsBySpec),
- replaceStatistics);
+ replaceStatistics,
+ partitionOptions == null
+ ? null
+ : partitionOptions.subList(offset, end));
+ offset = end;
}
return null;
},
"create partitions");
}
+ private void validatePartitionOptions(
+ @Nullable List> partitionOptions,
+ List> partitions) {
+ if (partitionOptions == null) {
+ return;
+ }
+ checkArgument(
+ partitionOptions.size() == partitions.size(),
+ "Partition options for table %s must align with all %s partition specs.",
+ identifier.getFullName(),
+ partitions.size());
+ Set> uniqueSpecs = capacityFor(partitions.size());
+ for (int i = 0; i < partitionOptions.size(); i++) {
+ Map options = partitionOptions.get(i);
+ checkArgument(options != null, "Partition options must not contain null maps.");
+ checkArgument(
+ options.entrySet().stream()
+ .noneMatch(entry -> entry.getKey() == null || entry.getValue() == null),
+ "Partition options must not contain null keys or values.");
+ checkArgument(
+ partitions.get(i) != null && uniqueSpecs.add(partitions.get(i)),
+ "Partition specs must be non-null and unique when partition options are provided.");
+ }
+ }
+
/**
* Indexes the reported statistics by the partition they describe, rejecting any that describes
* a partition this call does not register: a spec typo would otherwise account for nothing, or
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..1425cd538ee5 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,45 @@ private OptionalLong rowCount(List entries) {
private List, Path>> toSpecsAndPaths(
List partitions, boolean onlyValueInPath) {
+ if (partitions.stream()
+ .noneMatch(
+ partition ->
+ FormatTablePartitionPathResolver.customLocation(partition)
+ != 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, FormatTablePartitionPathResolver.customLocation(partition));
+ 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..d7169cfdd538 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,136 @@ 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 (FormatTablePartitionPathResolver.customLocation(partition) != 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(),
+ FormatTablePartitionPathResolver.customLocation(partition)));
+ }
+
private List publishMessage(TwoPhaseCommitMessage message) {
try {
message.getCommitter().commit(fileIO);
@@ -436,7 +575,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 +814,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 +1210,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 (FormatTablePartitionPathResolver.customLocation(partition) != 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 +1237,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 (FormatTablePartitionPathResolver.customLocation(partition) != 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..0c74250e410a 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,7 +66,7 @@ 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);
}
/**
@@ -78,7 +78,7 @@ default void createPartitions(List> partitions, boolean igno
* holds or add to it, and is ignored when {@code statistics} is null. A field reported as
* unknown says nothing about itself and leaves the stored one as it was, so a measurement that
* could not take a number does not erase the last one that could. Reporting never unregisters a
- * partition.
+ * partition. {@code partitionOptions}, when present, align with {@code partitions} by position.
*
* 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
@@ -88,7 +88,8 @@ void createPartitions(
List> partitions,
boolean ignoreIfExists,
@Nullable List statistics,
- boolean replaceStatistics);
+ boolean replaceStatistics,
+ @Nullable List> partitionOptions);
/** 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..6c5c6327e230
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionPathResolver.java
@@ -0,0 +1,507 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.partition.Partition;
+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 and validates catalog-managed Format Table partition locations. */
+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;
+ }
+
+ @Nullable
+ static String customLocation(Partition partition) {
+ Map options = partition.options();
+ if (options == null || !options.containsKey(CoreOptions.PATH.key())) {
+ return null;
+ }
+ String location = options.get(CoreOptions.PATH.key());
+ if (location == null) {
+ throw new IllegalStateException("Partition path option must not be null.");
+ }
+ return location;
+ }
+
+ 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..44c4f787aff0
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidator.java
@@ -0,0 +1,67 @@
+/*
+ * 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,
+ FormatTablePartitionPathResolver.customLocation(partition));
+ 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..3de7bbf73f16 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
@@ -364,7 +364,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 +383,40 @@ 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 testCreatePartitionsWithOptionsForwardsAndInvalidatesPartitionCache()
+ 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");
+ Map options = new HashMap<>();
+ options.put(CoreOptions.PATH.key(), "file:/archive/dt=20260717");
+ options.put("owner", "data-platform");
+ List> partitionOptions = singletonList(options);
+ 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, partitionOptions);
+
+ Mockito.verify(wrapped)
+ .createPartitions(
+ identifier, singletonList(spec), true, null, false, partitionOptions);
+ 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..63fa0b478181 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
@@ -18,12 +18,14 @@
package org.apache.paimon.catalog;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.partition.PartitionStatistics;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -53,9 +55,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 +69,26 @@ 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 testCreatePartitionsCarriesOptionsToTheWrappedCatalog() throws Exception {
+ Catalog wrapped = mock(Catalog.class);
+ Catalog delegating = new TestDelegateCatalog(wrapped);
+ Map spec = Collections.singletonMap("dt", "20260728");
+ List> specs = Collections.singletonList(spec);
+ Map options = new HashMap<>();
+ options.put(CoreOptions.PATH.key(), "file:/archive/dt=20260728");
+ options.put("owner", "data-platform");
+ List> partitionOptions = Collections.singletonList(options);
+
+ delegating.createPartitions(IDENTIFIER, specs, true, null, false, partitionOptions);
+
+ verify(wrapped).createPartitions(IDENTIFIER, specs, true, null, false, partitionOptions);
+ 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..aec8fb775e5d 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
@@ -51,9 +51,11 @@
import org.apache.paimon.rest.auth.DLFTokenLoader;
import org.apache.paimon.rest.auth.DLFTokenLoaderFactory;
import org.apache.paimon.rest.auth.RESTAuthParameter;
+import org.apache.paimon.rest.exceptions.AlreadyExistsException;
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;
@@ -78,12 +80,19 @@
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Paths;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
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;
@@ -92,6 +101,7 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.tuple;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -486,6 +496,347 @@ 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";
+ Map options = new HashMap<>();
+ options.put(CoreOptions.PATH.key(), requested);
+ options.put("owner", "data-platform");
+ 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(options));
+
+ assertThat(restCatalogServer.getReceivedHeaders(partitionsResource)).hasSize(1);
+ assertThat(onlyPartition(identifier).options())
+ .containsExactlyInAnyOrderEntriesOf(
+ ImmutableMap.of(
+ CoreOptions.PATH.key(), canonical, "owner", "data-platform"));
+ }
+
+ @Test
+ void testRenamePreservesCustomPartitionLocation() 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";
+ restCatalog.createPartitions(
+ source,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ partitionOptions(location));
+
+ restCatalog.renameTable(source, destination, false);
+
+ assertThat(restCatalog.listPartitions(destination))
+ .singleElement()
+ .satisfies(
+ partition -> {
+ assertThat(partition.spec()).isEqualTo(spec);
+ assertThat(customLocation(partition)).isEqualTo(location);
+ });
+ }
+
+ @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,
+ partitionOptions(
+ "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 RawCreatePartitionsRequest(
+ Arrays.asList(first, second),
+ partitionOptions("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,
+ partitionOptions(" ")),
+ restCatalog.api().authFunction()))
+ .isInstanceOf(BadRequestException.class)
+ .hasMessageContaining("Invalid custom partition location");
+ assertThatThrownBy(
+ () ->
+ client.post(
+ resource,
+ new CreatePartitionsRequest(
+ Arrays.asList(first, first),
+ true,
+ null,
+ null,
+ partitionOptions(
+ "file:/archive/one", "file:/archive/two")),
+ restCatalog.api().authFunction()))
+ .isInstanceOf(BadRequestException.class)
+ .hasMessageContaining("must not contain duplicates");
+ assertThat(restCatalog.listPartitions(identifier)).isEmpty();
+ }
+
+ @Test
+ void testPartitionOptionsRejectNullValuesBeforeMutation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ Map options = new HashMap<>();
+ options.put("owner", null);
+ String resource =
+ ResourcePaths.forCatalogProperties(restCatalog.api().options())
+ .partitions(identifier.getDatabaseName(), identifier.getObjectName());
+ HttpClient client = new HttpClient(restCatalogServer.getUrl());
+
+ assertThatThrownBy(
+ () ->
+ client.post(
+ resource,
+ new RawCreatePartitionsRequest(
+ Collections.singletonList(spec),
+ Collections.singletonList(options)),
+ restCatalog.api().authFunction()))
+ .isInstanceOf(BadRequestException.class)
+ .hasMessageContaining("null keys or values");
+ 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,
+ partitionOptions(locations.toArray(new String[0]))))
+ .isInstanceOf(IllegalArgumentException.class);
+ assertThat(restCatalog.listPartitions(identifier)).isEmpty();
+ }
+ }
+ }
+
+ @Test
+ void testExistingPartitionRejectsADifferentCustomLocationWithoutMutation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ String originalLocation = "file:/archive/original";
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ partitionOptions(originalLocation));
+
+ assertThatThrownBy(
+ () ->
+ restCatalog
+ .api()
+ .createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ partitionOptions("file:/archive/different")))
+ .isInstanceOf(AlreadyExistsException.class)
+ .hasMessageContaining("different location");
+
+ assertThat(restCatalog.listPartitions(identifier))
+ .extracting(Partition::spec, MockRESTCatalogTest::customLocation)
+ .containsExactly(tuple(spec, originalLocation));
+ }
+
+ @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,
+ partitionOptions(sharedLocation));
+ return null;
+ });
+ Future second =
+ executor.submit(
+ () -> {
+ start.await();
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(secondSpec),
+ true,
+ null,
+ false,
+ partitionOptions(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(customLocation(stored.get(0))).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.setPartitionOptionsCreateSupported(false);
+ restCatalogServer.clearReceivedHeaders();
+
+ assertThatThrownBy(
+ () ->
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ partitionOptions("file:/archive/dt=20260717")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("does not support partition options");
+
+ assertThat(restCatalogServer.getReceivedHeaders(resource)).hasSize(1);
+ assertThat(restCatalog.listPartitions(identifier)).isEmpty();
+ }
+
+ @Test
+ void testEmptyPartitionOptionsDoNotRequireProviderSupport() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ Map spec = Collections.singletonMap("dt", "20260717");
+ restCatalogServer.setPartitionOptionsCreateSupported(false);
+
+ restCatalog.createPartitions(
+ identifier,
+ Collections.singletonList(spec),
+ true,
+ null,
+ false,
+ Collections.singletonList(Collections.emptyMap()));
+
+ assertThat(onlyPartition(identifier).spec()).isEqualTo(spec);
+ assertThat(onlyPartition(identifier).options()).isNull();
+ }
+
+ @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,
+ partitionOptions("FILE:///archive//%64t%3D20260717/", null)),
+ restCatalog.api().authFunction());
+
+ assertThat(restCatalog.listPartitions(identifier))
+ .extracting(Partition::spec, MockRESTCatalogTest::customLocation)
+ .containsExactlyInAnyOrder(
+ tuple(first, "file:/archive/dt=20260717"), tuple(second, null));
+ }
+
@Test
void testPartitionManagerSurvivesSerialization() throws Exception {
Identifier identifier = createFormatTableWithCatalogManagedPartitions();
@@ -507,12 +858,17 @@ void testReportedPartitionStatisticsAreStoredAndReadBack() throws Exception {
Identifier identifier = createFormatTableWithCatalogManagedPartitions();
Map spec = Collections.singletonMap("dt", "20260717");
List> specs = Collections.singletonList(spec);
+ String location = "file:/archive/dt=20260717";
+ Map options = new HashMap<>();
+ options.put(CoreOptions.PATH.key(), location);
+ options.put("owner", "data-platform");
FormatTablePartitionManager partitionManager =
((FormatTable) restCatalog.getTable(identifier)).partitionManager();
assertThat(partitionManager).isNotNull();
// A registration on its own measures nothing, so everything starts out unknown.
- restCatalog.createPartitions(identifier, specs);
+ restCatalog.createPartitions(
+ identifier, specs, true, null, false, Collections.singletonList(options));
Partition registered = onlyPartition(identifier);
assertThat(PartitionStatistics.isKnown(registered.recordCount())).isFalse();
assertThat(PartitionStatistics.isKnown(registered.fileCount())).isFalse();
@@ -526,7 +882,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 +892,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 +909,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 +923,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,11 +940,13 @@ void testReportedPartitionStatisticsAreStoredAndReadBack() throws Exception {
PartitionStatistics.UNKNOWN,
PartitionStatistics.UNKNOWN,
-1)),
- true);
+ true,
+ null);
assertStatistics(identifier, 5L, 900L, 1L, 700L);
// Reporting never registers or unregisters anything.
assertThat(restCatalog.listPartitions(identifier)).hasSize(1);
+ assertThat(onlyPartition(identifier).options()).isEqualTo(options);
}
@Test
@@ -606,7 +968,8 @@ void testStatisticsOfAnUnstoredPartitionAreDropped() throws Exception {
3L,
1000L,
-1)),
- false);
+ false,
+ null);
assertThat(restCatalog.listPartitions(identifier))
.extracting(Partition::spec)
@@ -628,7 +991,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);
@@ -644,6 +1008,21 @@ private Partition onlyPartition(Identifier identifier) throws Exception {
return partitions.get(0);
}
+ private static List> partitionOptions(String... locations) {
+ List> options = new ArrayList<>(locations.length);
+ for (String location : locations) {
+ options.add(
+ location == null
+ ? Collections.emptyMap()
+ : Collections.singletonMap(CoreOptions.PATH.key(), location));
+ }
+ return options;
+ }
+
+ private static String customLocation(Partition partition) {
+ return partition.options() == null ? null : partition.options().get(CoreOptions.PATH.key());
+ }
+
private void assertStatistics(
Identifier identifier,
long recordCount,
@@ -1052,6 +1431,34 @@ private RESTCatalog initCatalogUtil(
return new RESTCatalog(CatalogContext.create(options));
}
+ private static class RawCreatePartitionsRequest implements RESTRequest {
+
+ private final List> partitionSpecs;
+ private final List> partitionOptions;
+
+ private RawCreatePartitionsRequest(
+ List> partitionSpecs,
+ List> partitionOptions) {
+ this.partitionSpecs = partitionSpecs;
+ this.partitionOptions = partitionOptions;
+ }
+
+ @JsonGetter("partitionSpecs")
+ public List> getPartitionSpecs() {
+ return partitionSpecs;
+ }
+
+ @JsonGetter("partitionOptions")
+ public List> getPartitionOptions() {
+ return partitionOptions;
+ }
+
+ @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..9641cf02f0d8 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,13 +59,16 @@
import org.junit.Test;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
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 +340,67 @@ public void createPartitionsRequestParseTest() throws Exception {
assertNull(defaultRequest.replaceStatistics());
}
+ @Test
+ public void createPartitionsRequestPreservesOptionsTest() throws Exception {
+ String json =
+ "{\"partitionSpecs\":[{\"dt\":\"20260901\"},{\"dt\":\"20260902\"}],"
+ + "\"partitionOptions\":[{},"
+ + "{\"path\":\"oss://archive-bucket/table/dt=20260902\","
+ + "\"owner\":\"data-platform\"}]}";
+
+ CreatePartitionsRequest request = RESTApi.fromJson(json, CreatePartitionsRequest.class);
+ Map serialized = RESTApi.fromJson(RESTApi.toJson(request), Map.class);
+ Map customOptions = new HashMap<>();
+ customOptions.put("path", "oss://archive-bucket/table/dt=20260902");
+ customOptions.put("owner", "data-platform");
+
+ assertEquals(
+ Arrays.asList(Collections.emptyMap(), customOptions),
+ serialized.get("partitionOptions"));
+ }
+
+ @Test
+ public void createPartitionsRequestRejectsNullOptionMapsTest() {
+ List> specs =
+ Arrays.asList(
+ Collections.singletonMap("dt", "20260901"),
+ Collections.singletonMap("dt", "20260902"));
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new CreatePartitionsRequest(
+ specs,
+ true,
+ null,
+ null,
+ Arrays.asList(null, Collections.emptyMap())));
+
+ Map nullValue = new HashMap<>();
+ nullValue.put("owner", null);
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new CreatePartitionsRequest(
+ specs,
+ true,
+ null,
+ null,
+ Arrays.asList(Collections.emptyMap(), nullValue)));
+
+ Map nullKey = new HashMap<>();
+ nullKey.put(null, "value");
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ new CreatePartitionsRequest(
+ specs,
+ true,
+ null,
+ null,
+ Arrays.asList(Collections.emptyMap(), nullKey)));
+ }
+
@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..a7036a9ee97f
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogPartitionSupport.java
@@ -0,0 +1,189 @@
+/*
+ * 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.partition.PartitionUtils;
+import org.apache.paimon.rest.requests.CreatePartitionsRequest;
+import org.apache.paimon.rest.responses.ErrorResponse;
+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.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.paimon.CoreOptions.PATH;
+
+/** Helpers for validating partition options in the mock REST catalog. */
+final class RESTCatalogPartitionSupport {
+
+ private RESTCatalogPartitionSupport() {}
+
+ @Nullable
+ static List> canonicalizeRequestedOptions(
+ CreatePartitionsRequest request,
+ CatalogContext catalogContext,
+ boolean optionCreateSupported) {
+ List> options = request.getPartitionOptions();
+ if (options == null) {
+ return null;
+ }
+ List> specs = request.getPartitionSpecs();
+ if (specs == null || options.size() != specs.size()) {
+ throw new IllegalArgumentException(
+ "partitionOptions must contain exactly one entry for every partition spec.");
+ }
+ Set> uniqueSpecs = new HashSet<>();
+ boolean hasOptions = false;
+ for (int i = 0; i < options.size(); i++) {
+ if (specs.get(i) == null || !uniqueSpecs.add(specs.get(i))) {
+ throw new IllegalArgumentException(
+ "partitionSpecs must not contain duplicates when partitionOptions is present.");
+ }
+ Map partitionOptions = options.get(i);
+ if (partitionOptions == null) {
+ throw new IllegalArgumentException("partitionOptions must not contain null maps.");
+ }
+ if (partitionOptions.entrySet().stream()
+ .anyMatch(entry -> entry.getKey() == null || entry.getValue() == null)) {
+ throw new IllegalArgumentException(
+ "partitionOptions must not contain null keys or values.");
+ }
+ hasOptions |= !partitionOptions.isEmpty();
+ }
+ if (!hasOptions) {
+ return null;
+ }
+ if (!optionCreateSupported) {
+ throw new UnsupportedOperationException(
+ "This REST provider does not support partition options.");
+ }
+ List> canonical = new ArrayList<>(options.size());
+ for (int i = 0; i < options.size(); i++) {
+ Map partitionOptions = options.get(i);
+ Map copied = new HashMap<>(partitionOptions);
+ String location = copied.get(PATH.key());
+ if (location != null) {
+ copied.put(
+ PATH.key(),
+ FormatTablePartitionPathResolver.canonicalizeCustomLocation(
+ location, catalogContext)
+ .toString());
+ }
+ canonical.add(copied);
+ }
+ return canonical;
+ }
+
+ static Optional> conflictingLocation(
+ List stored,
+ List> requestedSpecs,
+ @Nullable List> requestedOptions) {
+ if (requestedOptions == null) {
+ return Optional.empty();
+ }
+ Map, Partition> storedBySpec = new HashMap<>();
+ for (Partition partition : stored) {
+ storedBySpec.put(partition.spec(), partition);
+ }
+ for (int i = 0; i < requestedSpecs.size(); i++) {
+ Map spec = requestedSpecs.get(i);
+ Partition existing = storedBySpec.get(spec);
+ String requestedLocation = requestedOptions.get(i).get(PATH.key());
+ if (existing != null
+ && requestedLocation != null
+ && !Objects.equals(customLocation(existing), requestedLocation)) {
+ return Optional.of(spec);
+ }
+ }
+ return Optional.empty();
+ }
+
+ static ErrorResponse conflictingLocationError(Map spec) {
+ String partitionName = PartitionUtils.buildPartitionName(spec);
+ return new ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_PARTITION,
+ partitionName,
+ String.format(
+ "Partition %s already exists at a different location.", partitionName),
+ 409);
+ }
+
+ static Partition newPartition(Map spec, @Nullable Map options) {
+ return new Partition(
+ spec,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN,
+ PartitionStatistics.UNKNOWN_TOTAL_BUCKETS,
+ false,
+ null,
+ null,
+ null,
+ null,
+ options);
+ }
+
+ static void validateFormatTablePartitionLocations(
+ List partitions,
+ TableMetadata metadata,
+ String tableName,
+ CatalogContext catalogContext) {
+ if (partitions.stream().noneMatch(partition -> customLocation(partition) != 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);
+ }
+ }
+
+ @Nullable
+ private static String customLocation(Partition partition) {
+ return partition.options() == null ? null : partition.options().get(PATH.key());
+ }
+}
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..c94d53f4aafd 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,7 +253,7 @@ 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<>();
@@ -268,10 +269,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 partitionOptionsCreateSupported = true;
public RESTCatalogServer(
String dataPath, AuthProvider authProvider, ConfigResponse config, String warehouse) {
@@ -340,6 +343,10 @@ public void setPartitionListingSupported(boolean partitionListingSupported) {
this.partitionListingSupported = partitionListingSupported;
}
+ public void setPartitionOptionsCreateSupported(boolean partitionOptionsCreateSupported) {
+ this.partitionOptionsCreateSupported = partitionOptionsCreateSupported;
+ }
+
public void clearReceivedListPartitionsByFilterRequests() {
receivedListPartitionsByFilterRequests.clear();
}
@@ -433,7 +440,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();
@@ -635,8 +642,19 @@ && isTableByIdRequest(request.getPath())) {
|| isListPartitionsByFilter)) {
return mockResponse(new ErrorResponse(null, null, "", 501), 501);
} else if (isDropPartitions) {
- return dropPartitionsHandle(restAuthParameter.data(), identifier);
+ synchronized (tableLifecycleLocks.lock(identifier.getFullName())) {
+ return dropPartitionsHandle(restAuthParameter.data(), identifier);
+ }
} else if (isPartitions) {
+ if ("POST".equals(restAuthParameter.method())) {
+ synchronized (tableLifecycleLocks.lock(identifier.getFullName())) {
+ return partitionsApiHandle(
+ restAuthParameter.method(),
+ restAuthParameter.data(),
+ parameters,
+ identifier);
+ }
+ }
return partitionsApiHandle(
restAuthParameter.method(),
restAuthParameter.data(),
@@ -682,7 +700,9 @@ && 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);
@@ -2152,6 +2172,7 @@ private MockResponse renameTableHandle(String data) throws Exception {
current.isExternal());
tableMetadataStore.remove(fromTable.getFullName(), current);
tableMetadataStore.put(toTable.getFullName(), renamedMetadata);
+ renamePartitionState(fromTable, toTable);
permissionStore.renameTable(fromTable, toTable);
}
}
@@ -2161,6 +2182,16 @@ private MockResponse renameTableHandle(String data) throws Exception {
return new MockResponse().setResponseCode(200);
}
+ private void renamePartitionState(Identifier source, Identifier destination) {
+ String sourceName = source.getFullName();
+ String destinationName = destination.getFullName();
+ tablePartitionsStore.remove(destinationName);
+ List partitions = tablePartitionsStore.remove(sourceName);
+ if (partitions != null) {
+ tablePartitionsStore.put(destinationName, partitions);
+ }
+ }
+
private MockResponse partitionsApiHandle(
String method, String data, Map parameters, Identifier tableIdentifier)
throws Exception {
@@ -2185,9 +2216,18 @@ private MockResponse partitionsApiHandle(
return generateFinalListPartitionsResponse(parameters, partitions);
case "POST":
CreatePartitionsRequest request = parseRequest(data, CreatePartitionsRequest.class);
+ List> requestedOptions =
+ RESTCatalogPartitionSupport.canonicalizeRequestedOptions(
+ request, catalogContext, partitionOptionsCreateSupported);
+ String tableName = tableIdentifier.getFullName();
+ TableMetadata tableMetadata = tableMetadataStore.get(tableName);
+ if (tableMetadata == null) {
+ throw new Catalog.TableNotExistException(tableIdentifier);
+ }
List storedPartitions =
- tablePartitionsStore.computeIfAbsent(
- tableIdentifier.getFullName(), ignored -> new ArrayList<>());
+ new ArrayList<>(
+ tablePartitionsStore.getOrDefault(
+ tableName, Collections.emptyList()));
Set> existingSpecs =
storedPartitions.stream().map(Partition::spec).collect(Collectors.toSet());
if (!request.ignoreIfExists()) {
@@ -2209,30 +2249,42 @@ private MockResponse partitionsApiHandle(
return mockResponse(response, 409);
}
}
+ Optional> conflictingLocation =
+ RESTCatalogPartitionSupport.conflictingLocation(
+ storedPartitions, request.getPartitionSpecs(), requestedOptions);
+ if (conflictingLocation.isPresent()) {
+ return mockResponse(
+ RESTCatalogPartitionSupport.conflictingLocationError(
+ conflictingLocation.get()),
+ 409);
+ }
List