listPartitionsByNames(
throw new TableNoPermissionException(identifier, e);
} catch (NotImplementedException e) {
Table table = getTable(identifier);
- if (isCatalogManagedFormatTable(table)) {
+ if (hasCatalogManagedPartitions(table)) {
throw e;
}
return listPartitionsFromFileSystem(table, partitions);
@@ -1323,9 +1334,13 @@ RESTApi api() {
return api;
}
- private static boolean isCatalogManagedFormatTable(Table table) {
- return table instanceof FormatTable
- && new CoreOptions(table.options()).partitionedTableInMetastore();
+ /**
+ * Whether the catalog owns this table's partitions, which is exactly whether loading it
+ * produced a partition manager. Reading the option instead would be a second answer to the same
+ * question, and the two can disagree on a table built outside the catalog.
+ */
+ private static boolean hasCatalogManagedPartitions(Table table) {
+ return table instanceof FormatTable && ((FormatTable) table).partitionManager() != null;
}
private FileIO fileIOForData(Path path, Identifier identifier) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
index 9a832176caaf..acb487e2573b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
@@ -18,9 +18,12 @@
package org.apache.paimon.table;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.annotation.Experimental;
import org.apache.paimon.annotation.Public;
import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogUtils;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.manifest.IndexManifestEntry;
@@ -29,6 +32,7 @@
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.format.FormatBatchWriteBuilder;
import org.apache.paimon.table.format.FormatReadBuilder;
+import org.apache.paimon.table.format.FormatTablePartitionManager;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.StreamWriteBuilder;
import org.apache.paimon.table.source.BatchVectorSearchBuilder;
@@ -56,8 +60,9 @@
* operations on this table allow for reading or writing to these files, facilitating the retrieval
* of existing data and the addition of new files.
*
- * Partitioned file format table just like the standard hive format. Partitions are discovered
- * and inferred based on directory structure.
+ *
A partitioned file format table uses the standard Hive directory layout. By default,
+ * partitions are discovered from that layout. Format Tables with catalog-managed partitions use
+ * catalog metadata for partition visibility while retaining the same physical layout.
*
* @since 0.9.0
*/
@@ -75,6 +80,17 @@ public interface FormatTable extends Table {
CatalogContext catalogContext();
+ /**
+ * The catalog partition registrations of this table, or null when its partitions are discovered
+ * from the filesystem. When non-null, catalog registration decides partition visibility: a
+ * partition directory that is not registered is not visible to scans.
+ */
+ @Experimental
+ @Nullable
+ default FormatTablePartitionManager partitionManager() {
+ return null;
+ }
+
/** Currently supported formats. */
enum Format {
ORC,
@@ -115,6 +131,7 @@ class Builder {
private Map options;
@Nullable private String comment;
private CatalogContext catalogContext;
+ @Nullable private FormatTablePartitionManager partitionManager;
public Builder fileIO(FileIO fileIO) {
this.fileIO = fileIO;
@@ -161,6 +178,12 @@ public Builder catalogContext(CatalogContext catalogContext) {
return this;
}
+ @Experimental
+ public Builder partitionManager(@Nullable FormatTablePartitionManager partitionManager) {
+ this.partitionManager = partitionManager;
+ return this;
+ }
+
public FormatTable build() {
return new FormatTableImpl(
fileIO,
@@ -171,7 +194,8 @@ public FormatTable build() {
format,
options,
comment,
- catalogContext);
+ catalogContext,
+ partitionManager);
}
}
@@ -189,6 +213,7 @@ class FormatTableImpl implements FormatTable {
private final Map options;
@Nullable private final String comment;
private CatalogContext catalogContext;
+ @Nullable private final FormatTablePartitionManager partitionManager;
public FormatTableImpl(
FileIO fileIO,
@@ -199,7 +224,8 @@ public FormatTableImpl(
Format format,
Map options,
@Nullable String comment,
- CatalogContext catalogContext) {
+ CatalogContext catalogContext,
+ @Nullable FormatTablePartitionManager partitionManager) {
this.fileIO = fileIO;
this.identifier = identifier;
this.rowType = rowType;
@@ -209,6 +235,7 @@ public FormatTableImpl(
this.options = options;
this.comment = comment;
this.catalogContext = catalogContext;
+ this.partitionManager = partitionManager;
}
@Override
@@ -265,6 +292,30 @@ public FileIO fileIO() {
public FormatTable copy(Map dynamicOptions) {
Map newOptions = new HashMap<>(options);
newOptions.putAll(dynamicOptions);
+
+ CoreOptions coreOptions = CoreOptions.fromMap(options);
+ CoreOptions copiedCoreOptions = CoreOptions.fromMap(newOptions);
+ boolean hasCatalogManagedPartitions = partitionManager != null;
+ boolean persistedPartitionsFromCatalog = coreOptions.partitionedTableInMetastore();
+ boolean dynamicPartitionsFromCatalog = copiedCoreOptions.partitionedTableInMetastore();
+ if (persistedPartitionsFromCatalog != dynamicPartitionsFromCatalog) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Dynamic option '%s' cannot change where a Format Table's partitions come from.",
+ CoreOptions.METASTORE_PARTITIONED_TABLE.key()));
+ }
+ if (hasCatalogManagedPartitions
+ && coreOptions.formatTablePartitionOnlyValueInPath()
+ != copiedCoreOptions.formatTablePartitionOnlyValueInPath()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Dynamic option '%s' cannot change the physical partition layout of a Format Table with catalog-managed partitions.",
+ CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key()));
+ }
+ if (hasCatalogManagedPartitions) {
+ CatalogUtils.validateCatalogManagedPartitionOptions(newOptions);
+ }
+
return new FormatTableImpl(
fileIO,
identifier,
@@ -274,7 +325,8 @@ public FormatTable copy(Map dynamicOptions) {
format,
newOptions,
comment,
- catalogContext);
+ catalogContext,
+ partitionManager);
}
@Override
@@ -302,6 +354,12 @@ public FullTextSearchBuilder newFullTextSearchBuilder() {
public CatalogContext catalogContext() {
return this.catalogContext;
}
+
+ @Override
+ @Nullable
+ public FormatTablePartitionManager partitionManager() {
+ return partitionManager;
+ }
}
@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
new file mode 100644
index 000000000000..7085135f17ff
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java
@@ -0,0 +1,200 @@
+/*
+ * 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.PagedList;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogLoader;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.utils.FunctionWithException;
+import org.apache.paimon.utils.PartitionPathUtils;
+import org.apache.paimon.utils.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/**
+ * A {@link FormatTablePartitionManager} reading and writing partitions through a {@link Catalog}.
+ * One catalog is created per operation and closed when it ends, so the table stays serializable and
+ * no client outlives the call that needed it.
+ */
+class CatalogFormatTablePartitionManager implements FormatTablePartitionManager {
+
+ private static final long serialVersionUID = 1L;
+
+ /** Bounds one request: catalog services cap the partitions a single call may carry. */
+ private static final int REQUEST_SIZE = 1000;
+
+ private final Identifier identifier;
+ private final List partitionKeys;
+ private final CatalogLoader catalogLoader;
+
+ CatalogFormatTablePartitionManager(
+ Identifier identifier, List partitionKeys, CatalogLoader catalogLoader) {
+ this.identifier = identifier;
+ // Copied: the caller's list must not be able to change what counts as a leading prefix.
+ this.partitionKeys = Collections.unmodifiableList(new ArrayList<>(partitionKeys));
+ this.catalogLoader = catalogLoader;
+ }
+
+ @Override
+ public List listPartitions(Map prefix) {
+ LinkedHashMap ordered = orderedPrefix(partitionKeys, prefix);
+ checkArgument(
+ ordered.size() == prefix.size(),
+ "Partition prefix %s is not a leading prefix of partition keys %s of table %s.",
+ prefix,
+ partitionKeys,
+ identifier.getFullName());
+ String pattern = PartitionPathUtils.buildPartitionNamePrefixPattern(partitionKeys, ordered);
+ return execute(
+ catalog -> {
+ List partitions = new ArrayList<>();
+ Set seenPageTokens = new HashSet<>();
+ String pageToken = null;
+ do {
+ PagedList page =
+ catalog.listPartitionsPaged(
+ identifier, REQUEST_SIZE, pageToken, pattern);
+ if (page.getElements() != null) {
+ partitions.addAll(page.getElements());
+ }
+ pageToken = page.getNextPageToken();
+ if (StringUtils.isNotEmpty(pageToken) && !seenPageTokens.add(pageToken)) {
+ throw new IllegalStateException(
+ String.format(
+ "Catalog returned repeated partition page token '%s' for format table %s.",
+ pageToken, identifier.getFullName()));
+ }
+ } while (StringUtils.isNotEmpty(pageToken));
+ // A catalog may ignore the pattern, so the prefix is enforced here as well.
+ partitions.removeIf(partition -> !matchesPrefix(partition, prefix));
+ return Collections.unmodifiableList(partitions);
+ },
+ "list partitions");
+ }
+
+ @Override
+ public List listPartitionsByNames(List> partitions) {
+ if (partitions.isEmpty()) {
+ return Collections.emptyList();
+ }
+ return execute(
+ catalog -> {
+ List found = new ArrayList<>();
+ for (List> batch : batches(partitions)) {
+ found.addAll(catalog.listPartitionsByNames(identifier, batch));
+ }
+ return Collections.unmodifiableList(found);
+ },
+ "list partitions by names");
+ }
+
+ @Override
+ public void createPartitions(List> partitions, boolean ignoreIfExists) {
+ if (partitions.isEmpty()) {
+ return;
+ }
+ execute(
+ catalog -> {
+ if (!ignoreIfExists) {
+ // 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);
+ return null;
+ }
+ // An idempotent create is safe to split: a rerun converges from a partially
+ // applied batch.
+ for (List> batch : batches(partitions)) {
+ catalog.createPartitions(identifier, batch, true);
+ }
+ return null;
+ },
+ "create partitions");
+ }
+
+ @Override
+ public void dropPartitions(List> partitions) {
+ if (partitions.isEmpty()) {
+ return;
+ }
+ execute(
+ catalog -> {
+ // Dropping ignores missing partitions, so a partially applied batch converges
+ // on a rerun and may be split.
+ for (List> batch : batches(partitions)) {
+ catalog.dropPartitions(identifier, batch);
+ }
+ return null;
+ },
+ "drop partitions");
+ }
+
+ private static boolean matchesPrefix(Partition partition, Map prefix) {
+ for (Map.Entry entry : prefix.entrySet()) {
+ if (!entry.getValue().equals(partition.spec().get(entry.getKey()))) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static List>> batches(List> partitions) {
+ List>> batches = new ArrayList<>();
+ for (int start = 0; start < partitions.size(); start += REQUEST_SIZE) {
+ batches.add(
+ partitions.subList(start, Math.min(start + REQUEST_SIZE, partitions.size())));
+ }
+ return batches;
+ }
+
+ private T execute(FunctionWithException operation, String what) {
+ try (Catalog catalog = catalogLoader.load()) {
+ return operation.apply(catalog);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to %s of format table %s.", what, identifier.getFullName()),
+ e);
+ }
+ }
+
+ /** Order a prefix by the table partition keys, stopping at the first key it does not cover. */
+ private static LinkedHashMap orderedPrefix(
+ List partitionKeys, Map prefix) {
+ LinkedHashMap ordered = new LinkedHashMap<>();
+ for (String partitionKey : partitionKeys) {
+ if (!prefix.containsKey(partitionKey)) {
+ break;
+ }
+ ordered.put(partitionKey, prefix.get(partitionKey));
+ }
+ return ordered;
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
index 924e67dbdfdc..0b510c594051 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
@@ -87,7 +87,8 @@ public BatchTableCommit newCommit() {
Identifier.fromString(table.fullName()),
staticPartition,
syncHiveUri,
- table.catalogContext());
+ table.catalogContext(),
+ table.partitionManager());
}
@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 ed98c6ba50ff..c0f356fe226d 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
@@ -48,7 +48,6 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.StringJoiner;
import static org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition;
@@ -63,6 +62,7 @@ public class FormatTableCommit implements BatchTableCommit {
protected boolean overwrite = false;
private Catalog hiveCatalog;
private Identifier tableIdentifier;
+ @Nullable private final FormatTablePartitionManager partitionManager;
public FormatTableCommit(
String location,
@@ -73,7 +73,8 @@ public FormatTableCommit(
Identifier tableIdentifier,
@Nullable Map staticPartitions,
@Nullable String syncHiveUri,
- CatalogContext catalogContext) {
+ CatalogContext catalogContext,
+ @Nullable FormatTablePartitionManager partitionManager) {
this.location = location;
this.fileIO = fileIO;
this.formatTablePartitionOnlyValueInPath = formatTablePartitionOnlyValueInPath;
@@ -82,6 +83,7 @@ public FormatTableCommit(
this.overwrite = overwrite;
this.partitionKeys = partitionKeys;
this.tableIdentifier = tableIdentifier;
+ this.partitionManager = partitionManager;
if (syncHiveUri != null) {
try {
Options options = new Options();
@@ -143,7 +145,9 @@ public void commit(List commitMessages) {
for (TwoPhaseOutputStream.Committer committer : committers) {
committer.commit(this.fileIO);
- if (partitionKeys != null && !partitionKeys.isEmpty() && hiveCatalog != null) {
+ if (partitionKeys != null
+ && !partitionKeys.isEmpty()
+ && (hiveCatalog != null || partitionManager != null)) {
partitionSpecs.add(
extractPartitionSpecFromPath(
committer.targetPath().getParent(), partitionKeys));
@@ -152,6 +156,11 @@ public void commit(List commitMessages) {
for (TwoPhaseOutputStream.Committer committer : committers) {
committer.clean(this.fileIO);
}
+ if (partitionManager != null && !partitionSpecs.isEmpty()) {
+ // Concurrent writers may touch the same partition, so registration is an
+ // idempotent ADD rather than a strict create.
+ partitionManager.createPartitions(new ArrayList<>(partitionSpecs), true);
+ }
for (Map partitionSpec : partitionSpecs) {
if (hiveCatalog != null) {
try {
@@ -192,12 +201,25 @@ private Method getHiveCreatePartitionsInHmsMethod() throws NoSuchMethodException
private LinkedHashMap extractPartitionSpecFromPath(
Path partitionPath, List partitionKeys) {
- if (formatTablePartitionOnlyValueInPath) {
- return PartitionPathUtils.extractPartitionSpecFromPathOnlyValue(
- partitionPath, partitionKeys);
- } else {
- return PartitionPathUtils.extractPartitionSpecFromPath(partitionPath);
+ // Only the trailing partitionKeys.size() components are the spec: the table location
+ // itself may contain foreign 'k=v' segments that must not leak into what is registered.
+ LinkedHashMap partitionSpec =
+ formatTablePartitionOnlyValueInPath
+ ? PartitionPathUtils.extractPartitionSpecFromPathOnlyValue(
+ partitionPath, partitionKeys)
+ : PartitionPathUtils.extractPartitionSpecFromPath(
+ partitionPath, partitionKeys);
+ if (partitionSpec == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition path '%s' does not end in the %s partition directories of "
+ + "table %s declared by partition keys %s.",
+ partitionPath,
+ partitionKeys.size(),
+ tableIdentifier.getFullName(),
+ partitionKeys));
}
+ return partitionSpec;
}
private static Path buildPartitionPath(
@@ -208,20 +230,19 @@ private static Path buildPartitionPath(
if (partitionSpec.isEmpty() || partitionKeys.isEmpty()) {
throw new IllegalArgumentException("partitionSpec or partitionKeys is empty.");
}
- StringJoiner joiner = new StringJoiner("/");
+ LinkedHashMap orderedSpec = new LinkedHashMap<>();
for (int i = 0; i < partitionSpec.size(); i++) {
String key = partitionKeys.get(i);
if (partitionSpec.containsKey(key)) {
- if (formatTablePartitionOnlyValueInPath) {
- joiner.add(partitionSpec.get(key));
- } else {
- joiner.add(key + "=" + partitionSpec.get(key));
- }
+ orderedSpec.put(key, partitionSpec.get(key));
} else {
throw new RuntimeException("partitionSpec does not contain key: " + key);
}
}
- return new Path(location, joiner.toString());
+ return new Path(
+ location,
+ PartitionPathUtils.generatePartitionPathUtil(
+ orderedSpec, formatTablePartitionOnlyValueInPath));
}
@Override
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
new file mode 100644
index 000000000000..64b643fba26e
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java
@@ -0,0 +1,69 @@
+/*
+ * 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.annotation.Experimental;
+import org.apache.paimon.catalog.CatalogLoader;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.partition.Partition;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * The catalog partition registrations of a single Format Table with catalog-managed partitions.
+ *
+ * This manages registration metadata only: it never creates, deletes or resolves a partition
+ * directory. Partition values are raw values, never escaped directory names, and building
+ * directories from them — including keeping them inside the table — is the caller's job.
+ *
+ *
Paging, page tokens, name patterns and per-request batching are implementation details; every
+ * method here takes or returns its complete argument.
+ *
+ * @since 1.5
+ */
+@Experimental
+public interface FormatTablePartitionManager extends Serializable {
+
+ /**
+ * List every partition whose leading partition values match {@code prefix}, or all partitions
+ * when it is empty. The prefix must be a contiguous leading subset of the table's partition
+ * keys.
+ */
+ List listPartitions(Map prefix);
+
+ /** Return those of the given complete partition specs that are registered. */
+ List listPartitionsByNames(List> partitions);
+
+ /**
+ * Register partitions. With {@code ignoreIfExists=false} the whole batch is rejected when any
+ * partition already exists, so such a request is never split.
+ */
+ void createPartitions(List> partitions, boolean ignoreIfExists);
+
+ /** Unregister partitions. Metadata only; missing partitions are ignored. */
+ void dropPartitions(List> partitions);
+
+ /** Create a manager reading and writing partitions through a {@link CatalogLoader}. */
+ static FormatTablePartitionManager create(
+ Identifier identifier, List partitionKeys, CatalogLoader catalogLoader) {
+ return new CatalogFormatTablePartitionManager(identifier, partitionKeys, catalogLoader);
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
index cb0edefc82fc..44908749cad6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
@@ -32,6 +32,7 @@
import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.options.Options;
+import org.apache.paimon.partition.Partition;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.partition.PartitionPredicate.AndPartitionPredicate;
import org.apache.paimon.partition.PartitionPredicate.DefaultPartitionPredicate;
@@ -59,12 +60,14 @@
import javax.annotation.Nullable;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -81,9 +84,10 @@ public class FormatTableScan implements InnerTableScan {
private static final Logger LOG = LoggerFactory.getLogger(FormatTableScan.class);
- private final FormatTable table;
- private final CoreOptions coreOptions;
+ final FormatTable table;
+ final CoreOptions coreOptions;
@Nullable private PartitionPredicate partitionFilter;
+ @Nullable private final FormatTablePartitionManager partitionManager;
@Nullable private final Integer limit;
private final long targetSplitSize;
private final long openFileCost;
@@ -97,6 +101,7 @@ public FormatTableScan(
this.coreOptions = new CoreOptions(table.options());
this.partitionFilter = partitionFilter;
this.limit = limit;
+ this.partitionManager = table.partitionManager();
this.targetSplitSize = coreOptions.splitTargetSize();
this.openFileCost = coreOptions.splitOpenFileCost();
this.format = table.format();
@@ -115,6 +120,29 @@ public Plan plan() {
@Override
public List listPartitionEntries() {
+ if (partitionManager != null) {
+ List partitions = partitionManager.listPartitions(Collections.emptyMap());
+ if (partitions.isEmpty()) {
+ warnIfFilesystemPartitionsExist();
+ }
+ boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath();
+ List entries = new ArrayList<>(partitions.size());
+ Set> seen = new HashSet<>(partitions.size());
+ for (Partition partition : partitions) {
+ if (!seen.add(partition.spec())) {
+ continue;
+ }
+ entries.add(
+ new PartitionEntry(
+ toPartitionRow(normalizeSpec(partition.spec(), onlyValueInPath)),
+ partition.recordCount(),
+ partition.fileSizeInBytes(),
+ partition.fileCount(),
+ partition.lastFileCreationTime(),
+ partition.totalBuckets()));
+ }
+ return entries;
+ }
List, Path>> partition2Paths =
searchPartSpecAndPaths(
table.fileIO(),
@@ -142,7 +170,7 @@ public static boolean isDataFileName(String fileName) {
return fileName != null && !fileName.startsWith(".") && !fileName.startsWith("_");
}
- private BinaryRow toPartitionRow(LinkedHashMap partitionSpec) {
+ BinaryRow toPartitionRow(LinkedHashMap partitionSpec) {
RowType partitionType = table.partitionType();
GenericRow row =
convertSpecToInternalRow(partitionSpec, partitionType, table.defaultPartName());
@@ -160,7 +188,28 @@ public List splits() {
LinkedHashMap partitionSpec = pair.getKey();
BinaryRow partitionRow = toPartitionRow(partitionSpec);
if (partitionFilter == null || partitionFilter.test(partitionRow)) {
- splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow));
+ try {
+ splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow));
+ } catch (FileNotFoundException e) {
+ if (partitionManager == null) {
+ throw e;
+ }
+ // A registered partition without a directory reads as empty,
+ // matching Hive (e.g. ADD PARTITION before the first INSERT).
+ // Warn so a directory removed behind the catalog's back stays
+ // discoverable.
+ LOG.warn(
+ "Partition '{}' of format table {} is registered in "
+ + "the catalog but its directory '{}' does not "
+ + "exist; treating the partition as empty. If the "
+ + "directory was removed on purpose, drop the "
+ + "partition or repair the metadata, e.g. with "
+ + "MSCK REPAIR TABLE.",
+ PartitionPathUtils.generatePartitionName(
+ partitionSpec, false),
+ table.fullName(),
+ pair.getValue());
+ }
}
}
} else {
@@ -178,6 +227,16 @@ public List splits() {
}
List, Path>> findPartitions() {
+ if (partitionManager != null) {
+ Map prefix = leadingEqualityPrefix();
+ List partitions = partitionManager.listPartitions(prefix);
+ if (partitions.isEmpty() && prefix.isEmpty()) {
+ warnIfFilesystemPartitionsExist();
+ }
+ // The prefix is a coarse pre-filter; the full predicate is applied per partition in
+ // the plan, as it is for filesystem discovery.
+ return toSpecsAndPaths(partitions, coreOptions.formatTablePartitionOnlyValueInPath());
+ }
LOG.debug(
"Find partitions for format table {}, partition filter: {}",
table.name(),
@@ -222,6 +281,101 @@ List, Path>> findPartitions() {
}
}
+ /**
+ * Turn the partitions a catalog reports into the specs and directories to read. Catalog
+ * metadata is not trusted for path construction: a spec that cannot form a partition directory
+ * of this table is rejected rather than resolved to some other directory.
+ */
+ private List, Path>> toSpecsAndPaths(
+ List partitions, boolean onlyValueInPath) {
+ List, Path>> result = new ArrayList<>(partitions.size());
+ Path tablePath = new Path(table.location());
+ // Do not trust the catalog to be duplicate-free: a repeated spec would double every split
+ // of that partition and silently duplicate query results.
+ 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)));
+ }
+ }
+ return result;
+ }
+
+ /** Order a catalog spec by the table partition keys and check it can form a directory. */
+ private LinkedHashMap normalizeSpec(
+ @Nullable Map spec, boolean onlyValueInPath) {
+ List partitionKeys = table.partitionKeys();
+ if (spec == null
+ || spec.size() != partitionKeys.size()
+ || !spec.keySet().containsAll(partitionKeys)) {
+ throw corruptPartitionSpec(spec);
+ }
+ LinkedHashMap normalized = new LinkedHashMap<>();
+ for (String partitionKey : partitionKeys) {
+ String value = spec.get(partitionKey);
+ // In a value-only layout, '.' and '..' are complete path components and would resolve
+ // to a directory outside the table.
+ try {
+ PartitionPathUtils.validatePartitionValueForPath(value, onlyValueInPath);
+ } catch (IllegalArgumentException e) {
+ throw corruptPartitionSpec(spec);
+ }
+ normalized.put(partitionKey, value);
+ }
+ return normalized;
+ }
+
+ private IllegalStateException corruptPartitionSpec(@Nullable Map spec) {
+ return new IllegalStateException(
+ String.format(
+ "Catalog returned corrupt partition metadata %s for format table %s; "
+ + "expected exactly the partition keys %s with values usable as "
+ + "path components.",
+ spec, table.fullName(), table.partitionKeys()));
+ }
+
+ /**
+ * The leading equality prefix of the partition predicate, in partition-key order, pushed down
+ * to the partition catalog. Empty when there is no predicate or it does not start with
+ * equalities on the leading partition keys.
+ */
+ private Map leadingEqualityPrefix() {
+ Optional predicate = extractPartitionPredicate(partitionFilter);
+ if (!predicate.isPresent()) {
+ return Collections.emptyMap();
+ }
+ return extractLeadingEqualityPartitionSpecWhenOnlyAnd(
+ table.partitionKeys(), predicate.get(), table.partitionType());
+ }
+
+ /**
+ * Warn when the catalog knows no partitions but the table directory contains subdirectories:
+ * typically a table that predates catalog-managed partitions (or was written by a client that
+ * does not register partitions) and needs a metadata sync before its data becomes visible.
+ */
+ private void warnIfFilesystemPartitionsExist() {
+ try {
+ for (FileStatus status : table.fileIO().listStatus(new Path(table.location()))) {
+ if (status.isDir() && !status.getPath().getName().startsWith(".")) {
+ LOG.warn(
+ "Format table {} has no partitions registered in the catalog "
+ + "but its location {} contains directories. Data written "
+ + "before enabling catalog-managed partitions (or by clients "
+ + "that do not register partitions) is invisible until the "
+ + "partition metadata is synced, e.g. with MSCK REPAIR TABLE.",
+ table.fullName(),
+ table.location());
+ return;
+ }
+ }
+ } catch (IOException ignored) {
+ // Best-effort hint only; never fail or slow down the scan because of it.
+ }
+ }
+
protected static List, Path>> generatePartitions(
List partitionKeys,
RowType partitionType,
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
index fb7baf5b06b2..44ded39b6ab5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
@@ -109,13 +109,105 @@ public static String generatePartitionPathUtil(
suffixBuf.append(escapePathName(e.getKey()));
suffixBuf.append('=');
}
- suffixBuf.append(escapePathName(e.getValue()));
+ String value = e.getValue();
+ validatePartitionValueForPath(value, onlyValue);
+ suffixBuf.append(escapePathName(value));
i++;
}
suffixBuf.append(Path.SEPARATOR);
return suffixBuf.toString();
}
+ /**
+ * Generate a partition path without the trailing separator, e.g. {@code dt=20250101/hr=01}.
+ * This is the canonical partition name used when talking to a partition-managing catalog.
+ */
+ public static String generatePartitionName(
+ LinkedHashMap partitionSpec, boolean onlyValue) {
+ String path = generatePartitionPathUtil(partitionSpec, onlyValue);
+ return path.endsWith(Path.SEPARATOR)
+ ? path.substring(0, path.length() - Path.SEPARATOR.length())
+ : path;
+ }
+
+ /**
+ * Validate that a partition value is safe for the configured path layout. In a key-value
+ * layout, values such as {@code "."} are part of a component such as {@code "pt=."} and are
+ * safe. In a value-only layout, {@code "."} and {@code ".."} are complete path components and
+ * would resolve to a different directory.
+ */
+ public static void validatePartitionValueForPath(String value, boolean onlyValueInPath) {
+ if (value == null
+ || value.isEmpty()
+ || (onlyValueInPath && (".".equals(value) || "..".equals(value)))) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition value '%s' cannot be used as a partition path component.",
+ value));
+ }
+ }
+
+ /** Conservatively validate a value when the physical partition layout is unknown. */
+ public static void validatePartitionValueForPath(String value) {
+ validatePartitionValueForPath(value, true);
+ }
+
+ /** Validate every value of a partition spec for the configured path layout. */
+ public static void validatePartitionSpecForPath(
+ Map partitionSpec, boolean onlyValueInPath) {
+ for (String value : partitionSpec.values()) {
+ validatePartitionValueForPath(value, onlyValueInPath);
+ }
+ }
+
+ /** Conservatively validate a spec when the physical partition layout is unknown. */
+ public static void validatePartitionSpecForPath(Map partitionSpec) {
+ validatePartitionSpecForPath(partitionSpec, true);
+ }
+
+ /**
+ * Build the partition-name prefix pattern pushed down to a partition-managing catalog from the
+ * leading equality prefix of a partition predicate.
+ *
+ * Pattern contract (shared by every engine talking to the catalog): partition names are the
+ * escaped {@code key=value} form joined by {@code '/'}; {@code '%'} is the only wildcard and
+ * there is no escape sequence for it ({@code '_'} stays a literal). A complete spec matches the
+ * exact partition name; an incomplete prefix is suffixed with {@code '%'}.
+ *
+ *
Returns {@code null} whenever pushdown must be skipped and the caller should list all
+ * partitions instead: the equality prefix is empty, a prefix value is blank, or the escaped
+ * prefix contains a literal {@code '%'} that the contract cannot express.
+ */
+ @Nullable
+ public static String buildPartitionNamePrefixPattern(
+ List partitionKeys, Map equalityPrefix) {
+ if (equalityPrefix.isEmpty()) {
+ return null;
+ }
+ LinkedHashMap orderedPrefix = new LinkedHashMap<>();
+ for (String partitionKey : partitionKeys) {
+ if (!equalityPrefix.containsKey(partitionKey)) {
+ break;
+ }
+ String value = equalityPrefix.get(partitionKey);
+ if (StringUtils.isNullOrWhitespaceOnly(value)) {
+ return null;
+ }
+ orderedPrefix.put(partitionKey, value);
+ }
+ if (orderedPrefix.isEmpty()) {
+ return null;
+ }
+ String escapedPrefix = generatePartitionPath(orderedPrefix);
+ if (escapedPrefix.indexOf('%') >= 0) {
+ return null;
+ }
+ if (orderedPrefix.size() == partitionKeys.size()) {
+ return escapedPrefix.substring(0, escapedPrefix.length() - 1);
+ }
+ return escapedPrefix + '%';
+ }
+
public static List generatePartitionPaths(
List> partitions, RowType partitionType) {
return partitions.stream()
@@ -265,12 +357,53 @@ public static LinkedHashMap extractPartitionSpecFromPath(Path cu
return fullPartSpec;
}
+ /**
+ * Extract exactly the trailing {@code key=value} components for the declared partition keys, or
+ * null when the path does not end in them. Values are unescaped, so they are the raw partition
+ * values.
+ */
+ @Nullable
+ public static LinkedHashMap extractPartitionSpecFromPath(
+ Path currPath, List partitionKeys) {
+ String[] values = new String[partitionKeys.size()];
+ Path current = currPath;
+ for (int i = partitionKeys.size() - 1; i >= 0; i--) {
+ if (current == null) {
+ return null;
+ }
+ Matcher matcher = PARTITION_NAME_PATTERN.matcher(current.getName());
+ if (!matcher.matches()
+ || !partitionKeys.get(i).equals(unescapePathName(matcher.group(1)))) {
+ return null;
+ }
+ values[i] = unescapePathName(matcher.group(2));
+ current = current.getParent();
+ }
+
+ LinkedHashMap spec = new LinkedHashMap<>();
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ spec.put(partitionKeys.get(i), values[i]);
+ }
+ return spec;
+ }
+
+ @Nullable
public static LinkedHashMap extractPartitionSpecFromPathOnlyValue(
Path currPath, List partitionKeys) {
LinkedHashMap fullPartSpec = new LinkedHashMap<>();
String[] split = currPath.toString().split(Path.SEPARATOR);
+ if (split.length < partitionKeys.size()) {
+ return null;
+ }
for (int i = 0; i < partitionKeys.size(); i++) {
- fullPartSpec.put(partitionKeys.get(i), split[split.length - partitionKeys.size() + i]);
+ // Unescape the directory component so the extracted value is the RAW partition value,
+ // consistent with the key=value branch (extractPartitionSpecFromPath) and with the
+ // values the write path registers into a partition-managing catalog. Without this,
+ // directories containing escaped characters (e.g. a%3Ab) would round-trip to a
+ // different value than the one registered (a:b).
+ fullPartSpec.put(
+ partitionKeys.get(i),
+ unescapePathName(split[split.length - partitionKeys.size() + i]));
}
return fullPartSpec;
}
@@ -330,8 +463,9 @@ public static List, Path>> searchPartSpecAndP
part.getPath(), partitionKeys),
part.getPath()));
} else {
- LinkedHashMap spec = extractPartitionSpecFromPath(part.getPath());
- if (spec.size() != partitionKeys.size()) {
+ LinkedHashMap spec =
+ extractPartitionSpecFromPath(part.getPath(), partitionKeys);
+ if (spec == null) {
// illegal path, for example: /path/to/table/tmp/unknown, path without "="
continue;
}
@@ -360,8 +494,17 @@ private static FileStatus[] getFileStatusRecurse(
GenericRow values =
partitionType == null ? null : new GenericRow(partitionType.getFieldCount());
+ FileStatus fileStatus;
+ try {
+ fileStatus = fileIO.getFileStatus(path);
+ } catch (FileNotFoundException e) {
+ // A missing root simply means the table has no partitions yet.
+ return new FileStatus[0];
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to list files in " + path, e);
+ }
+
try {
- FileStatus fileStatus = fileIO.getFileStatus(path);
// Skip partition levels already fixed by the scan-path prefix.
int levelOffset = partitionKeys.size() - expectLevel;
listStatusRecursively(
@@ -378,9 +521,10 @@ private static FileStatus[] getFileStatusRecurse(
defaultPartValue,
levelOffset,
values);
- } catch (FileNotFoundException e) {
- return new FileStatus[0];
} catch (IOException e) {
+ // Never degrade a mid-scan failure into an empty listing: callers diff this result
+ // against partition metadata and an incomplete listing would deregister partitions
+ // that still exist.
throw new RuntimeException("Failed to list files in " + path, e);
}
@@ -412,7 +556,15 @@ private static void listStatusRecursively(
}
if (fileStatus.isDir()) {
- for (FileStatus stat : fileIO.listStatus(fileStatus.getPath())) {
+ FileStatus[] children;
+ try {
+ children = fileIO.listStatus(fileStatus.getPath());
+ } catch (FileNotFoundException e) {
+ // The directory vanished after the parent listed it: the partitions beneath it
+ // are gone, skipping just this subtree keeps the rest of the listing complete.
+ return;
+ }
+ for (FileStatus stat : children) {
int partitionKeyIndex = levelOffset + level;
String partitionKey = partitionKeys.get(partitionKeyIndex);
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java
new file mode 100644
index 000000000000..3ffb6a1ecbde
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.catalog;
+
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.paimon.CoreOptions.FILE_FORMAT;
+import static org.apache.paimon.CoreOptions.FORMAT_TABLE_IMPLEMENTATION;
+import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE;
+import static org.apache.paimon.CoreOptions.PATH;
+import static org.apache.paimon.CoreOptions.TYPE;
+import static org.apache.paimon.TableType.FORMAT_TABLE;
+import static org.apache.paimon.catalog.CatalogUtils.loadTable;
+import static org.apache.paimon.catalog.CatalogUtils.validateCreateTable;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link CatalogUtils}. */
+class CatalogUtilsTest {
+
+ @Test
+ void testRejectEngineImplementationForCatalogManagedPartitionsOnCreate() {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option(TYPE.key(), FORMAT_TABLE.toString())
+ .option(METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(FORMAT_TABLE_IMPLEMENTATION.key(), "engine")
+ .build();
+
+ // The combination is rejected where catalog-managed partitions are actually served.
+ assertThatThrownBy(
+ () ->
+ CatalogUtils.validateCatalogManagedFormatTablePartitions(
+ Identifier.create("db", "t"), schema.options(), false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key())
+ .hasMessageContaining(FORMAT_TABLE_IMPLEMENTATION.key());
+
+ // Elsewhere it is not this validation's business: the option is inert there.
+ validateCreateTable(schema, false);
+ }
+
+ @Test
+ void testEngineImplementationCannotHaveCatalogManagedPartitions() {
+ // Asking for catalog-managed partitions where they cannot be served is an error, not a
+ // table that silently reads its partitions from somewhere else.
+ assertThatThrownBy(() -> loadFormatTableRequestingCatalogManagedPartitions("engine", true))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key())
+ .hasMessageContaining(FORMAT_TABLE_IMPLEMENTATION.key());
+ }
+
+ @Test
+ void testOutsideRestCatalogPartitionsFromCatalogAreRejected() {
+ // The option names one source only, so a catalog that cannot serve it says so rather than
+ // reading the table's partitions from somewhere the option did not ask for.
+ assertThatThrownBy(() -> loadFormatTableRequestingCatalogManagedPartitions("paimon", false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key())
+ .hasMessageContaining("REST catalog");
+ }
+
+ @Test
+ void testLoadCatalogManagedPartitionsFromRestCatalog() throws Exception {
+ Table table = loadFormatTableRequestingCatalogManagedPartitions("paimon", true);
+ assertThat(table).isInstanceOf(FormatTable.class);
+ FormatTable formatTable = (FormatTable) table;
+ assertThat(
+ new org.apache.paimon.CoreOptions(formatTable.options())
+ .partitionedTableInMetastore())
+ .isTrue();
+ assertThat(formatTable.partitionManager()).isNotNull();
+ }
+
+ private static Table loadFormatTableRequestingCatalogManagedPartitions(
+ String formatTableImplementation, boolean isRestCatalog) throws Exception {
+ Identifier identifier =
+ Identifier.create("catalog_partition_db", "catalog_partition_table");
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .option(TYPE.key(), FORMAT_TABLE.toString())
+ .option(METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(FORMAT_TABLE_IMPLEMENTATION.key(), formatTableImplementation)
+ .option(FILE_FORMAT.key(), "parquet")
+ .option(
+ PATH.key(),
+ "file:///tmp/catalog_partition_db.db/catalog_partition_table")
+ .build();
+ TableMetadata metadata =
+ new TableMetadata(
+ TableSchema.create(0, schema), false, "catalog-partition-table-id");
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.catalogLoader()).thenReturn(() -> catalog);
+
+ return loadTable(
+ catalog,
+ identifier,
+ path -> new LocalFileIO(),
+ path -> new LocalFileIO(),
+ ignored -> metadata,
+ null,
+ null,
+ null,
+ isRestCatalog);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
index 04f67c018d1b..094ede3847ab 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
@@ -18,9 +18,12 @@
package org.apache.paimon.catalog;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.TableType;
import org.apache.paimon.fs.Path;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
@@ -77,4 +80,33 @@ public void testAlterDatabase() throws Exception {
false))
.isInstanceOf(UnsupportedOperationException.class);
}
+
+ @Test
+ public void testPartitionsFromCatalogAreRejectedOutsideRestCatalog() throws Exception {
+ String database = "rest_partition_source_db";
+ Identifier identifier = Identifier.create(database, "rest_partition_source_table");
+ catalog.createDatabase(database, false);
+ // Write the schema file directly to simulate a format table carrying the REST-only
+ // partition source in a filesystem catalog.
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString())
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .option(CoreOptions.METASTORE_PARTITIONED_TABLE.key(), "true")
+ .build();
+ Path tablePath =
+ ((FileSystemCatalog) DelegateCatalog.rootCatalog(catalog))
+ .getTableLocation(identifier);
+ new SchemaManager(fileIO, tablePath).createTable(schema);
+
+ // The option names one thing only, so a catalog that cannot serve it says so rather than
+ // reading the table's partitions from somewhere the option did not ask for.
+ assertThatThrownBy(() -> catalog.getTable(identifier))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(CoreOptions.METASTORE_PARTITIONED_TABLE.key())
+ .hasMessageContaining("REST catalog");
+ }
}
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 b4af3194b72f..6cc8f7042018 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
@@ -26,6 +26,8 @@
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.catalog.TableQueryAuthResult;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.FieldRef;
@@ -46,8 +48,11 @@
import org.apache.paimon.rest.exceptions.NotImplementedException;
import org.apache.paimon.rest.responses.ConfigResponse;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.format.FormatTablePartitionManager;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.InstantiationUtil;
import org.apache.paimon.utils.JsonSerdeUtil;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
@@ -216,8 +221,8 @@ void testCreateTableDefaultOptions() throws Exception {
}
@Test
- void testManagedFormatTablePagedPartitionListingDoesNotFallback() throws Exception {
- Identifier identifier = createManagedFormatTable();
+ void testCatalogManagedPagedPartitionListingDoesNotFallback() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
restCatalogServer.setPartitionListingSupported(false);
assertThatThrownBy(() -> restCatalog.listPartitionsPaged(identifier, null, null, null))
@@ -225,8 +230,8 @@ void testManagedFormatTablePagedPartitionListingDoesNotFallback() throws Excepti
}
@Test
- void testManagedFormatTablePartitionListingByNamesDoesNotFallback() throws Exception {
- Identifier identifier = createManagedFormatTable();
+ void testCatalogManagedPartitionListingByNamesDoesNotFallback() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
restCatalogServer.setPartitionListingSupported(false);
assertThatThrownBy(
@@ -239,15 +244,97 @@ void testManagedFormatTablePartitionListingByNamesDoesNotFallback() throws Excep
}
@Test
- void testManagedFormatTablePartitionListingDoesNotFallback() throws Exception {
- Identifier identifier = createManagedFormatTable();
+ void testCatalogManagedPartitionListingDoesNotFallback() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
restCatalogServer.setPartitionListingSupported(false);
assertThatThrownBy(() -> restCatalog.listPartitions(identifier))
.isInstanceOf(NotImplementedException.class);
}
- private Identifier createManagedFormatTable() throws Exception {
+ @Test
+ void testCatalogManagedPartitionListingReflectsCatalogMutationsImmediately() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ FormatTable table = (FormatTable) restCatalog.getTable(identifier);
+ FormatTablePartitionManager partitionManager = table.partitionManager();
+ assertThat(partitionManager).isNotNull();
+ assertThat(partitionManager.listPartitions(Collections.emptyMap())).isEmpty();
+ Map partition = Collections.singletonMap("dt", "20260717");
+
+ restCatalog.createPartitions(identifier, Collections.singletonList(partition));
+
+ // Listings are not cached, so a mutation through the catalog is visible to the next read.
+ assertThat(partitionManager.listPartitions(Collections.emptyMap()))
+ .extracting(org.apache.paimon.partition.Partition::spec)
+ .containsExactly(partition);
+
+ restCatalog.dropPartitions(identifier, Collections.singletonList(partition));
+
+ assertThat(partitionManager.listPartitions(Collections.emptyMap())).isEmpty();
+ }
+
+ @Test
+ void testPartitionManagerSurvivesSerialization() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ FormatTable table = (FormatTable) restCatalog.getTable(identifier);
+ Map partition = Collections.singletonMap("dt", "20260717");
+ restCatalog.createPartitions(identifier, Collections.singletonList(partition));
+
+ // A table travels to task processes; its partition catalog must rebuild its client there.
+ FormatTablePartitionManager roundTripped =
+ InstantiationUtil.clone(table.partitionManager());
+
+ assertThat(roundTripped.listPartitions(Collections.emptyMap()))
+ .extracting(org.apache.paimon.partition.Partition::spec)
+ .containsExactly(partition);
+ }
+
+ @Test
+ void testRejectCatalogManagedPartitionsOnExternalTableBeforeCreate() throws Exception {
+ Identifier identifier = Identifier.create("db1", "external_partitioned_format_table");
+ restCatalog.createDatabase(identifier.getDatabaseName(), true);
+ String externalPath = dataPath + "/external-partitioned-format-table";
+ Schema schema =
+ Schema.newBuilder()
+ .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString())
+ .option(CoreOptions.METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .option(CoreOptions.PATH.key(), externalPath)
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .build();
+
+ assertThatThrownBy(() -> restCatalog.createTable(identifier, schema, false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("internal table");
+ assertThat(restCatalog.listTables(identifier.getDatabaseName()))
+ .doesNotContain(identifier.getTableName());
+ assertThat(LocalFileIO.create().exists(new Path(externalPath))).isFalse();
+ }
+
+ @Test
+ void testRoundTrippedFormatTableReplacePassesClientValidation() throws Exception {
+ Identifier identifier = createFormatTableWithCatalogManagedPartitions();
+ FormatTable existing = (FormatTable) restCatalog.getTable(identifier);
+ Schema replacement =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .options(existing.options())
+ .build();
+ assertThat(replacement.options())
+ .containsEntry(CoreOptions.PATH.key(), existing.location());
+
+ // The mock service does not implement Format Table replacement. Reaching that response
+ // proves the REST client accepted the unchanged synthetic path from the loaded table.
+ assertThatThrownBy(() -> restCatalog.replaceTable(identifier, replacement, false))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("replaceTable does not support format tables");
+ }
+
+ private Identifier createFormatTableWithCatalogManagedPartitions() throws Exception {
Identifier identifier = Identifier.create("db1", "managed_partition_table");
restCatalog.createDatabase(identifier.getDatabaseName(), true);
restCatalog.createTable(
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index 504dc6fab6ec..0e386e8b10cb 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -76,6 +76,7 @@
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.Instant;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableSnapshot;
@@ -1535,8 +1536,8 @@ void testListPartitionsFromFile() throws Exception {
}
@Test
- void testCreatePartitionsForManagedFormatTable() throws Exception {
- Identifier identifier = Identifier.create("format_partition_db", "managed_table");
+ void testCreatePartitionsForCatalogManagedFormatTablePartitions() throws Exception {
+ Identifier identifier = Identifier.create("format_partition_db", "catalog_partition_table");
catalog.createDatabase(identifier.getDatabaseName(), true);
catalog.createTable(
identifier,
@@ -1576,8 +1577,9 @@ void testCreatePartitionsForManagedFormatTable() throws Exception {
}
@Test
- void testDropPartitionsForManagedFormatTable() throws Exception {
- Identifier identifier = Identifier.create("format_partition_db", "managed_drop_table");
+ void testDropPartitionsForCatalogManagedFormatTablePartitions() throws Exception {
+ Identifier identifier =
+ Identifier.create("format_partition_db", "catalog_partition_drop_table");
catalog.createDatabase(identifier.getDatabaseName(), true);
catalog.createTable(
identifier,
@@ -1639,6 +1641,70 @@ void testDropPartitionsLoadsNonManagedTableOnce() throws Exception {
Mockito.verify(catalogSpy, Mockito.times(1)).getTable(identifier);
}
+ @Test
+ void testCatalogManagedPartitionCommitAndScanMatchesFileSystemMode() throws Exception {
+ Identifier identifier =
+ Identifier.create("format_partition_db", "catalog_partition_scan_table");
+ catalog.createDatabase(identifier.getDatabaseName(), true);
+ catalog.createTable(
+ identifier,
+ Schema.newBuilder()
+ .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString())
+ .option(METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(CoreOptions.FILE_FORMAT.key(), "csv")
+ .column("id", DataTypes.INT())
+ .column("year", DataTypes.INT())
+ .column("month", DataTypes.INT())
+ .partitionKeys("year", "month")
+ .build(),
+ false);
+
+ FormatTable managedTable = (FormatTable) catalog.getTable(identifier);
+ BatchWriteBuilder writeBuilder = managedTable.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(1, 2024, 10));
+ write.write(GenericRow.of(2, 2025, 10));
+ write.write(GenericRow.of(3, 2025, 11));
+ commit.commit(write.prepareCommit());
+ }
+
+ List> registeredPartitions =
+ Arrays.asList(
+ ImmutableMap.of("year", "2024", "month", "10"),
+ ImmutableMap.of("year", "2025", "month", "10"),
+ ImmutableMap.of("year", "2025", "month", "11"));
+ assertThat(catalog.listPartitions(identifier).stream().map(Partition::spec))
+ .containsExactlyInAnyOrderElementsOf(registeredPartitions);
+ // The table carries the catalog partition metadata its scan reads from.
+ assertThat(managedTable.partitionManager()).isNotNull();
+
+ Map fileSystemOptions = new HashMap<>(managedTable.options());
+ fileSystemOptions.put(METASTORE_PARTITIONED_TABLE.key(), "false");
+ FormatTable fileSystemTable =
+ FormatTable.builder()
+ .fileIO(managedTable.fileIO())
+ .identifier(Identifier.create("format_partition_db", "filesystem_scan"))
+ .rowType(managedTable.rowType())
+ .partitionKeys(managedTable.partitionKeys())
+ .location(managedTable.location())
+ .format(managedTable.format())
+ .options(fileSystemOptions)
+ .catalogContext(managedTable.catalogContext())
+ .build();
+
+ Map partitionFilter = singletonMap("year", "2025");
+ List readFromCatalog = read(managedTable, null, null, partitionFilter, null);
+ // Assert the rows themselves first: comparing the two reads alone would also pass if
+ // both stopped returning anything.
+ assertThat(readFromCatalog)
+ .extracting(row -> row.getInt(0) + "," + row.getInt(1) + "," + row.getInt(2))
+ .containsExactlyInAnyOrder("2,2025,10", "3,2025,11");
+ assertThat(readFromCatalog)
+ .containsExactlyInAnyOrderElementsOf(
+ read(fileSystemTable, null, null, partitionFilter, null));
+ }
+
@Test
void testListPartitions() throws Exception {
innerTestListPartitions(true);
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java
new file mode 100644
index 000000000000..36064eea52f8
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java
@@ -0,0 +1,174 @@
+/*
+ * 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;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.table.format.FormatTablePartitionManager;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH;
+import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE;
+import static org.apache.paimon.CoreOptions.READ_BATCH_SIZE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Compatibility tests for the public {@link FormatTable} API. */
+class FormatTableCompatibilityTest {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testCopyRejectsCatalogManagedPartitionStateChange(boolean withCatalogManagedPartitions) {
+ FormatTable table = formatTable(Boolean.toString(withCatalogManagedPartitions));
+
+ assertThatThrownBy(
+ () ->
+ table.copy(
+ Collections.singletonMap(
+ METASTORE_PARTITIONED_TABLE.key(),
+ Boolean.toString(!withCatalogManagedPartitions))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key());
+ }
+
+ @Test
+ void testCopyAllowsSemanticallyEquivalentCatalogManagedPartitionOption() {
+ FormatTable table = formatTable("true");
+
+ FormatTable copied =
+ table.copy(Collections.singletonMap(METASTORE_PARTITIONED_TABLE.key(), "true"));
+
+ assertThat(copied).isNotSameAs(table);
+ assertThat(copied.options()).containsEntry(METASTORE_PARTITIONED_TABLE.key(), "true");
+ }
+
+ @Test
+ void testCopyRejectsChangingAbsentCatalogManagedPartitionDefault() {
+ FormatTable table = formatTable(Collections.emptyMap(), null);
+
+ assertThat(table.options()).doesNotContainKey(METASTORE_PARTITIONED_TABLE.key());
+ assertThatThrownBy(
+ () ->
+ table.copy(
+ Collections.singletonMap(
+ METASTORE_PARTITIONED_TABLE.key(), "true")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testCatalogManagedPartitionCopyRejectsPhysicalPartitionLayoutChange(
+ boolean onlyValueInPath) {
+ Map options = new LinkedHashMap<>();
+ options.put(METASTORE_PARTITIONED_TABLE.key(), "true");
+ options.put(
+ FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(), Boolean.toString(onlyValueInPath));
+ // The layout is only fixed for a table that actually has catalog-managed partitions.
+ FormatTable table =
+ formatTable(
+ options,
+ FormatTablePartitionManager.create(
+ Identifier.create(
+ "catalog_partition_db", "catalog_partition_table"),
+ Collections.singletonList("dt"),
+ () -> null));
+
+ assertThatThrownBy(
+ () ->
+ table.copy(
+ Collections.singletonMap(
+ FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(),
+ Boolean.toString(!onlyValueInPath))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testCopyPreservesPartitionModeAndCatalogProviderForUnrelatedOption(
+ boolean withCatalogManagedPartitions) {
+ Identifier identifier =
+ Identifier.create("catalog_partition_db", "catalog_partition_table");
+ FormatTablePartitionManager partitionManager =
+ FormatTablePartitionManager.create(
+ identifier, Collections.singletonList("dt"), () -> null);
+ Map options = new LinkedHashMap<>();
+ options.put(
+ METASTORE_PARTITIONED_TABLE.key(), Boolean.toString(withCatalogManagedPartitions));
+ FormatTable table = formatTable(options, partitionManager);
+
+ FormatTable copied = table.copy(Collections.singletonMap(READ_BATCH_SIZE.key(), "128"));
+
+ assertThat(copied).isNotSameAs(table);
+ assertThat(copied.options())
+ .containsEntry(READ_BATCH_SIZE.key(), "128")
+ .containsEntry(
+ METASTORE_PARTITIONED_TABLE.key(),
+ Boolean.toString(withCatalogManagedPartitions));
+ assertThat(
+ new org.apache.paimon.CoreOptions(copied.options())
+ .partitionedTableInMetastore())
+ .isEqualTo(withCatalogManagedPartitions);
+ assertThat(copied.partitionManager()).isSameAs(partitionManager);
+ assertThat(table.options())
+ .containsExactlyEntriesOf(options)
+ .doesNotContainKey(READ_BATCH_SIZE.key());
+ }
+
+ @Test
+ void testPartitionManagerIsOptionalForOtherImplementations() throws Exception {
+ // Implementations outside this repository do not have to know about catalog-managed
+ // partitions.
+ assertThat(FormatTable.class.getMethod("partitionManager").isDefault()).isTrue();
+ }
+
+ private static FormatTable formatTable(String partitionedInMetastore) {
+ return formatTable(
+ Collections.singletonMap(METASTORE_PARTITIONED_TABLE.key(), partitionedInMetastore),
+ null);
+ }
+
+ private static FormatTable formatTable(
+ Map options, FormatTablePartitionManager partitionManager) {
+ return FormatTable.builder()
+ .fileIO(LocalFileIO.create())
+ .identifier(Identifier.create("catalog_partition_db", "catalog_partition_table"))
+ .rowType(
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("dt", DataTypes.STRING())
+ .build())
+ .partitionKeys(Collections.singletonList("dt"))
+ .location("file:///warehouse/catalog_partition_table")
+ .format(FormatTable.Format.PARQUET)
+ .options(options)
+ .partitionManager(partitionManager)
+ .build();
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java
new file mode 100644
index 000000000000..d7a5d97e691a
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java
@@ -0,0 +1,417 @@
+/*
+ * 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.PagedList;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogLoader;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.utils.InstantiationUtil;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.catchThrowable;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link CatalogFormatTablePartitionManager}: paging, prefix pushdown, per-request
+ * batching and catalog lifecycle are transport details this adapter owns, and none of them are
+ * visible through {@link FormatTablePartitionManager}.
+ */
+class CatalogFormatTablePartitionManagerTest {
+
+ private static final Identifier IDENTIFIER =
+ Identifier.create("catalog_partition_db", "catalog_partition_table");
+
+ private static final List PARTITION_KEYS = Arrays.asList("year", "month");
+
+ /** One catalog request never carries more than this many partitions. */
+ private static final int REQUEST_SIZE = 1000;
+
+ /** Handed to the serializable loader, which cannot capture a test instance field. */
+ private static Catalog staticCatalog;
+
+ // ------------------------------------------------------------------------
+ // paging
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testListPartitionsFollowsEveryPage() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition first = partition("2025", "01");
+ Partition second = partition("2025", "02");
+ when(catalog.listPartitionsPaged(any(), any(), any(), any()))
+ .thenReturn(
+ new PagedList<>(Collections.singletonList(first), "page-2"),
+ new PagedList<>(Collections.singletonList(second), null));
+
+ List partitions =
+ partitionManager(catalog).listPartitions(Collections.emptyMap());
+
+ assertThat(partitions).containsExactly(first, second);
+ ArgumentCaptor pageTokens = ArgumentCaptor.forClass(String.class);
+ verify(catalog, times(2))
+ .listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), pageTokens.capture(), any());
+ assertThat(pageTokens.getAllValues()).containsExactly(null, "page-2");
+ }
+
+ @Test
+ void testEmptyNextPageTokenEndsPaging() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition only = partition("2025", "01");
+ when(catalog.listPartitionsPaged(any(), any(), any(), any()))
+ .thenReturn(new PagedList<>(Collections.singletonList(only), ""));
+
+ List partitions =
+ partitionManager(catalog).listPartitions(Collections.emptyMap());
+
+ assertThat(partitions).containsExactly(only);
+ verify(catalog, times(1)).listPartitionsPaged(any(), any(), any(), any());
+ }
+
+ @Test
+ void testRepeatedPageTokenFailsFast() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ // a -> b -> a: following the cycle would list the same pages forever.
+ when(catalog.listPartitionsPaged(any(), any(), any(), any()))
+ .thenReturn(
+ new PagedList<>(Collections.singletonList(partition("2025", "01")), "a"),
+ new PagedList<>(Collections.singletonList(partition("2025", "02")), "b"),
+ new PagedList<>(Collections.singletonList(partition("2025", "03")), "a"));
+
+ FormatTablePartitionManager partitionManager = partitionManager(catalog);
+
+ assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap()))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("repeated partition page token")
+ .hasMessageContaining("catalog_partition_db.catalog_partition_table");
+ }
+
+ // ------------------------------------------------------------------------
+ // prefix pushdown and client side filtering
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testLeadingPrefixIsPushedDownAsPattern() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(any(), any(), any(), any()))
+ .thenReturn(new PagedList<>(Collections.emptyList(), null));
+
+ partitionManager(catalog).listPartitions(Collections.singletonMap("year", "2025"));
+
+ verify(catalog)
+ .listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), isNull(), eq("year=2025/%"));
+ }
+
+ @Test
+ void testEmptyPrefixPushesDownNoPattern() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(any(), any(), any(), any()))
+ .thenReturn(new PagedList<>(Collections.emptyList(), null));
+
+ partitionManager(catalog).listPartitions(Collections.emptyMap());
+
+ verify(catalog).listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), isNull(), isNull());
+ }
+
+ @Test
+ void testPrefixIsEnforcedOnCatalogsThatIgnoreThePattern() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition matching = partition("2025", "01");
+ Partition other = partition("2024", "01");
+ when(catalog.listPartitionsPaged(any(), any(), any(), any()))
+ .thenReturn(new PagedList<>(Arrays.asList(matching, other), null));
+
+ List partitions =
+ partitionManager(catalog).listPartitions(Collections.singletonMap("year", "2025"));
+
+ assertThat(partitions).containsExactly(matching);
+ }
+
+ @Test
+ void testNonLeadingPrefixIsRejected() {
+ Catalog catalog = mock(Catalog.class);
+ FormatTablePartitionManager partitionManager = partitionManager(catalog);
+ // "month" without "year" cannot be expressed as a partition path prefix.
+ Map prefix = Collections.singletonMap("month", "10");
+
+ assertThatThrownBy(() -> partitionManager.listPartitions(prefix))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("leading prefix");
+ }
+
+ // ------------------------------------------------------------------------
+ // batching
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testIdempotentCreateIsSplitIntoRequests() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ List> specs = specs(2500);
+
+ partitionManager(catalog).createPartitions(specs, true);
+
+ List>> batches = capturedCreates(catalog, true, 3);
+ assertThat(batches).extracting(List::size).containsExactly(1000, 1000, 500);
+ assertThat(flatten(batches)).isEqualTo(specs);
+ }
+
+ @Test
+ void testStrictCreateStaysOneRequest() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ List> specs = specs(2500);
+
+ partitionManager(catalog).createPartitions(specs, false);
+
+ // Rejecting the whole batch when any partition exists only holds if it is never split.
+ List>> batches = capturedCreates(catalog, false, 1);
+ assertThat(batches.get(0)).isEqualTo(specs);
+ }
+
+ @Test
+ void testDropIsSplitIntoRequests() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ List> specs = specs(2500);
+
+ partitionManager(catalog).dropPartitions(specs);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor>> captor = ArgumentCaptor.forClass(List.class);
+ verify(catalog, times(3)).dropPartitions(eq(IDENTIFIER), captor.capture());
+ assertThat(captor.getAllValues()).extracting(List::size).containsExactly(1000, 1000, 500);
+ assertThat(flatten(captor.getAllValues())).isEqualTo(specs);
+ }
+
+ @Test
+ void testListByNamesIsSplitAndUnioned() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ List> specs = specs(2500);
+ Partition first = partition("2025", "0000");
+ Partition second = partition("2025", "1000");
+ Partition third = partition("2025", "2000");
+ when(catalog.listPartitionsByNames(any(), anyList()))
+ .thenReturn(
+ Collections.singletonList(first),
+ Collections.singletonList(second),
+ Collections.singletonList(third));
+
+ List found = partitionManager(catalog).listPartitionsByNames(specs);
+
+ assertThat(found).containsExactly(first, second, third);
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor>> captor = ArgumentCaptor.forClass(List.class);
+ verify(catalog, times(3)).listPartitionsByNames(eq(IDENTIFIER), captor.capture());
+ assertThat(captor.getAllValues()).extracting(List::size).containsExactly(1000, 1000, 500);
+ assertThat(flatten(captor.getAllValues())).isEqualTo(specs);
+ }
+
+ @Test
+ void testEmptyInputTouchesNoCatalog() {
+ Catalog createCatalog = mock(Catalog.class);
+ partitionManager(createCatalog).createPartitions(Collections.emptyList(), true);
+ verifyNoInteractions(createCatalog);
+
+ Catalog dropCatalog = mock(Catalog.class);
+ partitionManager(dropCatalog).dropPartitions(Collections.emptyList());
+ verifyNoInteractions(dropCatalog);
+
+ Catalog listCatalog = mock(Catalog.class);
+ assertThat(partitionManager(listCatalog).listPartitionsByNames(Collections.emptyList()))
+ .isEmpty();
+ verifyNoInteractions(listCatalog);
+ }
+
+ // ------------------------------------------------------------------------
+ // catalog lifecycle
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testEveryOperationClosesTheCatalog() throws Exception {
+ Catalog listCatalog = mock(Catalog.class);
+ when(listCatalog.listPartitionsPaged(any(), any(), any(), any()))
+ .thenReturn(new PagedList<>(Collections.emptyList(), null));
+ partitionManager(listCatalog).listPartitions(Collections.emptyMap());
+ verify(listCatalog).close();
+
+ Catalog byNamesCatalog = mock(Catalog.class);
+ when(byNamesCatalog.listPartitionsByNames(any(), anyList()))
+ .thenReturn(Collections.emptyList());
+ partitionManager(byNamesCatalog).listPartitionsByNames(specs(1));
+ verify(byNamesCatalog).close();
+
+ Catalog createCatalog = mock(Catalog.class);
+ partitionManager(createCatalog).createPartitions(specs(1), true);
+ verify(createCatalog).close();
+
+ Catalog dropCatalog = mock(Catalog.class);
+ partitionManager(dropCatalog).dropPartitions(specs(1));
+ verify(dropCatalog).close();
+ }
+
+ @Test
+ void testFailingOperationStillClosesTheCatalog() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(any(), any(), any(), any()))
+ .thenThrow(new RuntimeException("catalog unavailable"));
+ FormatTablePartitionManager partitionManager = partitionManager(catalog);
+
+ assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap()))
+ .isInstanceOf(RuntimeException.class);
+
+ verify(catalog).close();
+ }
+
+ @Test
+ void testRejectedPrefixNeverLoadsACatalog() {
+ Catalog catalog = mock(Catalog.class);
+ FormatTablePartitionManager partitionManager = partitionManager(catalog);
+ Map prefix = Collections.singletonMap("month", "10");
+
+ assertThatThrownBy(() -> partitionManager.listPartitions(prefix))
+ .isInstanceOf(IllegalArgumentException.class);
+
+ verifyNoInteractions(catalog);
+ }
+
+ // ------------------------------------------------------------------------
+ // error translation
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testCheckedExceptionIsWrappedWithTableContext() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Catalog.TableNotExistException notExist = new Catalog.TableNotExistException(IDENTIFIER);
+ when(catalog.listPartitionsPaged(any(), any(), any(), any())).thenThrow(notExist);
+ FormatTablePartitionManager partitionManager = partitionManager(catalog);
+
+ Throwable thrown =
+ catchThrowable(() -> partitionManager.listPartitions(Collections.emptyMap()));
+
+ assertThat(thrown)
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("list partitions")
+ .hasMessageContaining("catalog_partition_db.catalog_partition_table");
+ assertThat(thrown.getCause()).isSameAs(notExist);
+ }
+
+ @Test
+ void testRuntimeExceptionIsRethrownUnchanged() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ RuntimeException failure = new IllegalStateException("catalog unavailable");
+ doThrow(failure).when(catalog).createPartitions(any(), anyList(), anyBoolean());
+ FormatTablePartitionManager partitionManager = partitionManager(catalog);
+
+ Throwable thrown = catchThrowable(() -> partitionManager.createPartitions(specs(1), true));
+
+ assertThat(thrown).isSameAs(failure);
+ }
+
+ // ------------------------------------------------------------------------
+ // serializability
+ // ------------------------------------------------------------------------
+
+ @Test
+ void testSurvivesJavaSerialization() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition only = partition("2025", "01");
+ when(catalog.listPartitionsPaged(any(), any(), any(), any()))
+ .thenReturn(new PagedList<>(Collections.singletonList(only), null));
+ staticCatalog = catalog;
+
+ FormatTablePartitionManager partitionManager =
+ FormatTablePartitionManager.create(IDENTIFIER, PARTITION_KEYS, new TestLoader());
+ FormatTablePartitionManager cloned = InstantiationUtil.clone(partitionManager);
+
+ assertThat(cloned.listPartitions(Collections.emptyMap())).containsExactly(only);
+ verify(catalog).close();
+ }
+
+ // ------------------------------------------------------------------------
+ // helpers
+ // ------------------------------------------------------------------------
+
+ private static FormatTablePartitionManager partitionManager(Catalog catalog) {
+ return FormatTablePartitionManager.create(IDENTIFIER, PARTITION_KEYS, () -> catalog);
+ }
+
+ private static List>> capturedCreates(
+ Catalog catalog, boolean ignoreIfExists, int expectedRequests) throws Exception {
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor>> captor = ArgumentCaptor.forClass(List.class);
+ verify(catalog, times(expectedRequests))
+ .createPartitions(eq(IDENTIFIER), captor.capture(), eq(ignoreIfExists));
+ return captor.getAllValues();
+ }
+
+ private static List> flatten(List>> batches) {
+ return batches.stream().flatMap(List::stream).collect(Collectors.toList());
+ }
+
+ private static Partition partition(String year, String month) {
+ return new Partition(spec(year, month), 0, 0, 0, 0, -1, false);
+ }
+
+ private static Map spec(String year, String month) {
+ LinkedHashMap spec = new LinkedHashMap<>();
+ spec.put("year", year);
+ spec.put("month", month);
+ return spec;
+ }
+
+ private static List> specs(int count) {
+ List> specs = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ specs.add(spec("2025", String.format("%04d", i)));
+ }
+ return specs;
+ }
+
+ /** A {@link CatalogLoader} that survives serialization by holding no state of its own. */
+ private static class TestLoader implements CatalogLoader {
+
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Catalog load() {
+ return staticCatalog;
+ }
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
new file mode 100644
index 000000000000..da74ea32732f
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java
@@ -0,0 +1,451 @@
+/*
+ * 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.PagedList;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.PartitionPathUtils;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for Format Table scans whose partitions are catalog-managed. */
+class CatalogManagedPartitionScanTest {
+
+ private static final Identifier IDENTIFIER =
+ Identifier.create("catalog_partition_db", "catalog_partition_table");
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testLeadingPatternResidualFilterAndUnregisteredDirectory() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%")))
+ .thenReturn(
+ new PagedList<>(
+ Arrays.asList(partition("2025", "10"), partition("2025", "11")),
+ null));
+ TrackingLocalFileIO fileIO = new TrackingLocalFileIO();
+ Path tablePath = new Path(tempDir.toUri());
+ Path octoberFile = writeDataFile(fileIO, tablePath, "year=2025/month=10");
+ Path novemberFile = writeDataFile(fileIO, tablePath, "year=2025/month=11");
+ Path unregisteredFile = writeDataFile(fileIO, tablePath, "year=2025/month=12");
+ FormatTable table = createTable(fileIO, tablePath, partitionManager(catalog), false);
+ PredicateBuilder builder = new PredicateBuilder(table.partitionType());
+ Predicate predicate =
+ PredicateBuilder.and(builder.equal(0, 2025), builder.greaterThan(1, 10));
+ PartitionPredicate filter =
+ PartitionPredicate.fromPredicate(table.partitionType(), predicate);
+
+ FormatTableScan scan = new FormatTableScan(table, filter, null);
+ List plannedFiles = plannedFiles(scan.plan().splits());
+
+ assertThat(plannedFiles).containsExactly(novemberFile);
+ assertThat(plannedFiles).doesNotContain(octoberFile, unregisteredFile);
+ assertThat(fileIO.listedPaths).containsExactly(new Path(tablePath, "year=2025/month=11"));
+ }
+
+ @Test
+ void testListPartitionEntriesUsesCatalogVisibility() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition catalogPartition =
+ new Partition(partition("2025", "11").spec(), 11L, 22L, 3L, 44L, 5, false);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(catalogPartition), null));
+ TrackingLocalFileIO fileIO = new TrackingLocalFileIO();
+ Path tablePath = new Path(tempDir.toUri());
+ writeDataFile(fileIO, tablePath, "year=2025/month=11");
+ writeDataFile(fileIO, tablePath, "year=2025/month=12");
+ FormatTable table = createTable(fileIO, tablePath, partitionManager(catalog), false);
+
+ List entries =
+ new FormatTableScan(table, null, null).listPartitionEntries();
+
+ assertThat(entries)
+ .extracting(
+ entry -> entry.partition().getInt(0), entry -> entry.partition().getInt(1))
+ .containsExactly(org.assertj.core.groups.Tuple.tuple(2025, 11));
+ assertThat(entries.get(0).recordCount()).isEqualTo(11L);
+ assertThat(entries.get(0).fileSizeInBytes()).isEqualTo(22L);
+ assertThat(entries.get(0).fileCount()).isEqualTo(3L);
+ assertThat(entries.get(0).lastFileCreationTime()).isEqualTo(44L);
+ assertThat(entries.get(0).totalBuckets()).isEqualTo(5);
+ assertThat(fileIO.listedPaths).isEmpty();
+ assertThat(fileIO.statusListedPaths).isEmpty();
+ }
+
+ @Test
+ void testUnderscoreInPartitionNameRemainsLiteralPrefix() {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTable table =
+ createStringPartitionTable(
+ fileIO, tablePath, recordingCatalog(Collections.emptyList()));
+ Predicate predicate =
+ new PredicateBuilder(table.partitionType())
+ .equal(0, BinaryString.fromString("a_b"));
+ PartitionPredicate filter =
+ PartitionPredicate.fromPredicate(table.partitionType(), predicate);
+
+ List plannedFiles =
+ plannedFiles(new FormatTableScan(table, filter, null).plan().splits());
+
+ // '_' has no special meaning in the prefix contract; it must not widen the match.
+ assertThat(plannedFiles).isEmpty();
+ assertThat(requestedPrefixes).containsExactly(Collections.singletonMap("year", "a_b"));
+ }
+
+ @Test
+ void testValueOnlyPartitionPath() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(
+ new PagedList<>(Collections.singletonList(partition("2025", "11")), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path dataFile = writeDataFile(fileIO, tablePath, "2025/11");
+ FormatTable table = createTable(fileIO, tablePath, partitionManager(catalog), true);
+
+ List plannedFiles =
+ plannedFiles(new FormatTableScan(table, null, null).plan().splits());
+
+ assertThat(plannedFiles).containsExactly(dataFile);
+ }
+
+ @Test
+ void testDuplicateCatalogPartitionPlansSplitsOnce() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(
+ new PagedList<>(
+ Arrays.asList(partition("2025", "11"), partition("2025", "11")),
+ null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path dataFile = writeDataFile(fileIO, tablePath, "year=2025/month=11");
+ FormatTable table = createTable(fileIO, tablePath, partitionManager(catalog), false);
+
+ List plannedFiles =
+ plannedFiles(new FormatTableScan(table, null, null).plan().splits());
+
+ assertThat(plannedFiles).containsExactly(dataFile);
+ }
+
+ @Test
+ void testWhitespacePartitionValueIsVisible() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition whitespacePartition = partition(" ", "11");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(whitespacePartition), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path dataFile =
+ writeDataFile(
+ fileIO,
+ tablePath,
+ PartitionPathUtils.generatePartitionPath(
+ new LinkedHashMap<>(whitespacePartition.spec())));
+ FormatTable table =
+ createStringPartitionTable(fileIO, tablePath, partitionManager(catalog));
+
+ List plannedFiles =
+ plannedFiles(new FormatTableScan(table, null, null).plan().splits());
+
+ assertThat(plannedFiles).containsExactly(dataFile);
+ }
+
+ @Test
+ void testEmptyPartitionValueReportsCorruptMetadata() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition emptyValuePartition = partition("2025", "");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(emptyValuePartition), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTable table =
+ createStringPartitionTable(fileIO, tablePath, partitionManager(catalog));
+
+ assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("corrupt partition metadata")
+ .hasMessageContaining(IDENTIFIER.getFullName());
+ }
+
+ @Test
+ void testCatalogFailureDoesNotFallBackToFileSystem() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ RuntimeException catalogFailure =
+ new RuntimeException("Catalog partition listing unavailable");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenThrow(catalogFailure);
+ TrackingLocalFileIO fileIO = new TrackingLocalFileIO();
+ Path tablePath = new Path(tempDir.toUri());
+ writeDataFile(fileIO, tablePath, "year=2025/month=11");
+ FormatTable table = createTable(fileIO, tablePath, partitionManager(catalog), false);
+
+ assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits())
+ .isSameAs(catalogFailure);
+ assertThat(fileIO.listedPaths).isEmpty();
+ }
+
+ @Test
+ @DisplayName(
+ "treats a registered partition with a missing directory as an empty partition "
+ + "(requires real object-store validation)")
+ void testMissingCatalogRegisteredPartitionReadsAsEmpty() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(
+ new PagedList<>(Collections.singletonList(partition("2025", "11")), null));
+ Path tablePath = new Path(tempDir.toUri());
+ Path missingPath = new Path(tablePath, "year=2025/month=11");
+ LocalFileIO fileIO =
+ new LocalFileIO() {
+ @Override
+ public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
+ assertThat(path).isEqualTo(missingPath);
+ throw new FileNotFoundException(path.toString());
+ }
+ };
+ FormatTable table = createTable(fileIO, tablePath, partitionManager(catalog), false);
+
+ // A registered partition whose directory is missing (e.g. ADD PARTITION before the first
+ // INSERT) must not fail the whole scan; it reads as an empty partition, matching Hive.
+ List plannedFiles =
+ plannedFiles(new FormatTableScan(table, null, null).plan().splits());
+ assertThat(plannedFiles).isEmpty();
+ }
+
+ @Test
+ void testTraversalPartitionValueReportsCorruptMetadata() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition traversalPartition = partition("2025", "..");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(traversalPartition), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTable table =
+ createStringPartitionTable(fileIO, tablePath, partitionManager(catalog), true);
+
+ assertThatThrownBy(() -> new FormatTableScan(table, null, null).plan().splits())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("corrupt partition metadata")
+ .hasMessageContaining(IDENTIFIER.getFullName());
+ }
+
+ @Test
+ void testDotDotValueInKeyValueLayoutRemainsVisible() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition partition = partition("2025", "..");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(partition), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path dataFile = writeDataFile(fileIO, tablePath, "year=2025/month=..");
+ FormatTable table =
+ createStringPartitionTable(fileIO, tablePath, partitionManager(catalog));
+
+ List plannedFiles =
+ plannedFiles(new FormatTableScan(table, null, null).plan().splits());
+
+ assertThat(plannedFiles).containsExactly(dataFile);
+ }
+
+ private final List> requestedPrefixes = new ArrayList<>();
+
+ private FormatTablePartitionManager partitionManager(Catalog catalog) {
+ return FormatTablePartitionManager.create(
+ IDENTIFIER, Arrays.asList("year", "month"), () -> catalog);
+ }
+
+ /** Records the prefix the scan pushes down and answers from a fixed partition list. */
+ private FormatTablePartitionManager recordingCatalog(List partitions) {
+ List> prefixes = requestedPrefixes;
+ return new FormatTablePartitionManager() {
+ @Override
+ public List listPartitions(Map prefix) {
+ prefixes.add(new LinkedHashMap<>(prefix));
+ List matching = new ArrayList<>();
+ for (Partition partition : partitions) {
+ boolean matches = true;
+ for (Map.Entry entry : prefix.entrySet()) {
+ if (!entry.getValue().equals(partition.spec().get(entry.getKey()))) {
+ matches = false;
+ break;
+ }
+ }
+ if (matches) {
+ matching.add(partition);
+ }
+ }
+ return matching;
+ }
+
+ @Override
+ public List listPartitionsByNames(List> partitions) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void createPartitions(
+ List> partitions, boolean ignoreIfExists) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void dropPartitions(List> partitions) {
+ throw new UnsupportedOperationException();
+ }
+ };
+ }
+
+ private FormatTable createTable(
+ LocalFileIO fileIO,
+ Path tablePath,
+ FormatTablePartitionManager partitionManager,
+ boolean valueOnlyPath) {
+ RowType rowType =
+ RowType.builder()
+ .field("year", DataTypes.INT())
+ .field("month", DataTypes.INT())
+ .field("id", DataTypes.INT())
+ .build();
+ return FormatTable.builder()
+ .fileIO(fileIO)
+ .identifier(IDENTIFIER)
+ .rowType(rowType)
+ .partitionKeys(Arrays.asList("year", "month"))
+ .location(tablePath.toString())
+ .format(FormatTable.Format.CSV)
+ .options(
+ Collections.singletonMap(
+ FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(),
+ Boolean.toString(valueOnlyPath)))
+ .partitionManager(partitionManager)
+ .build();
+ }
+
+ private FormatTable createStringPartitionTable(
+ LocalFileIO fileIO, Path tablePath, FormatTablePartitionManager partitionManager) {
+ return createStringPartitionTable(fileIO, tablePath, partitionManager, false);
+ }
+
+ private FormatTable createStringPartitionTable(
+ LocalFileIO fileIO,
+ Path tablePath,
+ FormatTablePartitionManager partitionManager,
+ boolean valueOnlyPath) {
+ RowType rowType =
+ RowType.builder()
+ .field("year", DataTypes.STRING())
+ .field("month", DataTypes.STRING())
+ .field("id", DataTypes.INT())
+ .build();
+ return FormatTable.builder()
+ .fileIO(fileIO)
+ .identifier(IDENTIFIER)
+ .rowType(rowType)
+ .partitionKeys(Arrays.asList("year", "month"))
+ .location(tablePath.toString())
+ .format(FormatTable.Format.CSV)
+ .options(
+ Collections.singletonMap(
+ FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(),
+ Boolean.toString(valueOnlyPath)))
+ .partitionManager(partitionManager)
+ .build();
+ }
+
+ private Path writeDataFile(LocalFileIO fileIO, Path tablePath, String partitionPath)
+ throws IOException {
+ Path file = new Path(new Path(tablePath, partitionPath), "data.csv");
+ fileIO.mkdirs(file.getParent());
+ try (PositionOutputStream out = fileIO.newOutputStream(file, false)) {
+ out.write(1);
+ }
+ return file;
+ }
+
+ private static Partition partition(String year, String month) {
+ Map spec = new LinkedHashMap<>();
+ spec.put("year", year);
+ spec.put("month", month);
+ return new Partition(spec, 0, 0, 0, 0, -1, false);
+ }
+
+ private static List plannedFiles(List splits) {
+ return splits.stream()
+ .map(FormatDataSplit.class::cast)
+ .flatMap(split -> split.files().stream())
+ .map(FormatDataSplit.FileMeta::filePath)
+ .collect(Collectors.toList());
+ }
+
+ private static class TrackingLocalFileIO extends LocalFileIO {
+
+ private final List listedPaths = new ArrayList<>();
+ private final List statusListedPaths = new ArrayList<>();
+
+ @Override
+ public FileStatus[] listStatus(Path path) throws IOException {
+ statusListedPaths.add(path);
+ return super.listStatus(path);
+ }
+
+ @Override
+ public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
+ listedPaths.add(path);
+ return super.listFiles(path, recursive);
+ }
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
new file mode 100644
index 000000000000..c7f8c8dac1af
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
@@ -0,0 +1,247 @@
+/*
+ * 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.Identifier;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.RenamingTwoPhaseOutputStream;
+import org.apache.paimon.fs.TwoPhaseOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.utils.PartitionPathUtils;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.ArgumentCaptor;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link FormatTableCommit}. */
+class FormatTableCommitTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testPartitionRegistrationFailureDiscardsTheFilesItWrote() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path targetPath = new Path(tablePath, "year=2025/month=10/data-1.csv");
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+ outputStream.write(1);
+ TwoPhaseOutputStream.Committer committer = outputStream.closeForCommit();
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ RuntimeException registrationFailure =
+ new RuntimeException("Catalog partition registration unavailable");
+ doThrow(registrationFailure).when(partitionManager).createPartitions(anyList(), eq(true));
+
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ false,
+ false,
+ Identifier.create("catalog_partition_db", "catalog_partition_table"),
+ null,
+ null,
+ null,
+ partitionManager);
+ CommitMessage message = new TwoPhaseCommitMessage(committer);
+
+ assertThatThrownBy(() -> commit.commit(Collections.singletonList(message)))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseMessage("Catalog partition registration unavailable");
+
+ // A failed write leaves nothing behind, whichever step failed: rerunning it converges,
+ // and an idempotent registration makes a partition that was registered anyway harmless.
+ assertThat(fileIO.exists(targetPath)).isFalse();
+ verify(partitionManager).createPartitions(anyList(), eq(true));
+ }
+
+ @Test
+ void testFileCommitFailureStillDiscardsUncommittedFiles() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class);
+ doThrow(new IOException("data commit failed")).when(committer).commit(fileIO);
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ false,
+ false,
+ Identifier.create("catalog_partition_db", "catalog_partition_table"),
+ null,
+ null,
+ null,
+ partitionManager);
+ CommitMessage message = new TwoPhaseCommitMessage(committer);
+
+ assertThatThrownBy(() -> commit.commit(Collections.singletonList(message)))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseMessage("data commit failed");
+
+ verify(committer).discard(fileIO);
+ verify(partitionManager, never()).createPartitions(anyList(), eq(true));
+ }
+
+ @Test
+ void testRegistersRawPartitionValuesForEscapedPath() throws Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ LinkedHashMap rawSpec = new LinkedHashMap<>();
+ rawSpec.put("year", "2025");
+ rawSpec.put("month", "a b:c");
+ // The writer escapes partition values when building the directory layout.
+ String partitionDir = PartitionPathUtils.generatePartitionPathUtil(rawSpec, false);
+ assertThat(partitionDir).isEqualTo("year=2025/month=a b%3Ac/");
+
+ FormatTablePartitionManager partitionManager =
+ commitPartitionedFile(tablePath, false, partitionDir);
+
+ // The catalog must receive RAW values; readers re-escape them when probing directories.
+ assertThat(registeredSpec(partitionManager))
+ .containsExactly(entry("year", "2025"), entry("month", "a b:c"));
+ }
+
+ @Test
+ void testForeignKeyValueSegmentsInLocationDoNotLeakIntoSpec() throws Exception {
+ Path tablePath = new Path(new Path(tempDir.toUri()), "env=prod/warehouse/tbl");
+
+ FormatTablePartitionManager partitionManager =
+ commitPartitionedFile(tablePath, false, "year=2025/month=10");
+
+ assertThat(registeredSpec(partitionManager))
+ .containsExactly(entry("year", "2025"), entry("month", "10"));
+ }
+
+ @Test
+ void testValueOnlyPathUnderForeignKeyValueSegmentRegistersRawValues() throws Exception {
+ Path tablePath = new Path(new Path(tempDir.toUri()), "env=prod/warehouse/tbl");
+ LinkedHashMap rawSpec = new LinkedHashMap<>();
+ rawSpec.put("year", "2025");
+ rawSpec.put("month", "a:b");
+ String partitionDir = PartitionPathUtils.generatePartitionPathUtil(rawSpec, true);
+ assertThat(partitionDir).isEqualTo("2025/a%3Ab/");
+
+ FormatTablePartitionManager partitionManager =
+ commitPartitionedFile(tablePath, true, partitionDir);
+
+ assertThat(registeredSpec(partitionManager))
+ .containsExactly(entry("year", "2025"), entry("month", "a:b"));
+ }
+
+ @Test
+ void testPathNotMatchingThePartitionKeysFails() {
+ Path tablePath = new Path(tempDir.toUri());
+
+ // The message names the path and the declared keys, which is what tells a reader that
+ // 'day' is not where 'month' was expected.
+ assertThatThrownBy(() -> commitPartitionedFile(tablePath, false, "year=2025/day=10"))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseInstanceOf(IllegalArgumentException.class)
+ .rootCause()
+ .hasMessageContaining("year=2025/day=10")
+ .hasMessageContaining("catalog_partition_db.catalog_partition_table")
+ .hasMessageContaining("[year, month]");
+ }
+
+ @Test
+ void testValueOnlyStaticPartitionCannotEscapeTableLocation() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path parentPath = new Path(tempDir.toUri());
+ Path tablePath = new Path(parentPath, "table");
+ Path siblingPath = new Path(parentPath, "keep");
+ fileIO.mkdirs(tablePath);
+ fileIO.mkdirs(siblingPath);
+ Map staticPartition = Collections.singletonMap("year", "..");
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Collections.singletonList("year"),
+ fileIO,
+ true,
+ true,
+ Identifier.create("catalog_partition_db", "catalog_partition_table"),
+ staticPartition,
+ null,
+ null,
+ null);
+
+ assertThatThrownBy(() -> commit.commit(Collections.emptyList()))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseInstanceOf(IllegalArgumentException.class)
+ .hasRootCauseMessage(
+ "Partition value '..' cannot be used as a partition path component.");
+ assertThat(fileIO.exists(tablePath)).isTrue();
+ assertThat(fileIO.exists(siblingPath)).isTrue();
+ }
+
+ private FormatTablePartitionManager commitPartitionedFile(
+ Path tableLocation, boolean onlyValueInPath, String partitionDir) throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path targetPath = new Path(new Path(tableLocation, partitionDir), "data-1.csv");
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+ outputStream.write(1);
+ TwoPhaseOutputStream.Committer committer = outputStream.closeForCommit();
+ FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class);
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tableLocation.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ onlyValueInPath,
+ false,
+ Identifier.create("catalog_partition_db", "catalog_partition_table"),
+ null,
+ null,
+ null,
+ partitionManager);
+ commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer)));
+ return partitionManager;
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static Map registeredSpec(
+ FormatTablePartitionManager partitionManager) {
+ ArgumentCaptor>> captor =
+ ArgumentCaptor.forClass((Class) List.class);
+ verify(partitionManager).createPartitions(captor.capture(), eq(true));
+ assertThat(captor.getValue()).hasSize(1);
+ return captor.getValue().get(0);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
index a21b168f54ea..6b2952778003 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
@@ -19,17 +19,25 @@
package org.apache.paimon.utils;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.LinkedHashMap;
import static org.apache.paimon.utils.PartitionPathUtils.mightMatch;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
-/** Tests for {@link PartitionPathUtils#mightMatch}. */
+/** Tests for {@link PartitionPathUtils}. */
class PartitionPathUtilsTest {
private final RowType partitionType =
@@ -39,6 +47,46 @@ class PartitionPathUtilsTest {
.build();
private final PredicateBuilder builder = new PredicateBuilder(partitionType);
+ @ParameterizedTest
+ @ValueSource(strings = {".", ".."})
+ void testValueOnlyDotSegmentIsRejected(String rawValue) {
+ LinkedHashMap partitionSpec = new LinkedHashMap<>();
+ partitionSpec.put("pt", rawValue);
+
+ assertThatThrownBy(() -> PartitionPathUtils.generatePartitionPathUtil(partitionSpec, true))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(rawValue);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {".", ".."})
+ void testKeyedDotValueKeepsCompatibleSafeLayout(String rawValue) {
+ LinkedHashMap partitionSpec = new LinkedHashMap<>();
+ partitionSpec.put("pt", rawValue);
+
+ String partitionPath = PartitionPathUtils.generatePartitionPathUtil(partitionSpec, false);
+ Path resolvedPath = new Path(new Path("file:///warehouse/table"), partitionPath);
+
+ assertThat(partitionPath).isEqualTo("pt=" + rawValue + Path.SEPARATOR);
+ assertThat(PartitionPathUtils.extractPartitionSpecFromPath(resolvedPath))
+ .containsExactly(entry("pt", rawValue));
+ }
+
+ @Test
+ void testTrailingPartitionExtractionStopsAtTableBoundary() {
+ Path partitionPath = new Path("file:///warehouse/env=prod/table/dt=20260718/hh=10");
+
+ assertThat(
+ PartitionPathUtils.extractPartitionSpecFromPath(
+ partitionPath, Arrays.asList("dt", "hh")))
+ .containsExactly(entry("dt", "20260718"), entry("hh", "10"));
+ assertThat(
+ PartitionPathUtils.extractPartitionSpecFromPath(
+ new Path("file:///warehouse/dt=parent/table/wrong=20260718/hh=10"),
+ Arrays.asList("dt", "hh")))
+ .isNull();
+ }
+
@Test
void testNullPredicate() {
assertThat(mightMatch(null, 0, 0, GenericRow.of(2024, 5))).isTrue();