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..f15278a7fd36 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,31 @@ 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.formatTablePartitionsFromCatalog();
+ boolean dynamicPartitionsFromCatalog =
+ copiedCoreOptions.formatTablePartitionsFromCatalog();
+ if (persistedPartitionsFromCatalog != dynamicPartitionsFromCatalog) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Dynamic option '%s' cannot change where a Format Table's partitions come from.",
+ CoreOptions.FORMAT_TABLE_PARTITION_SOURCE.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 +326,8 @@ public FormatTable copy(Map dynamicOptions) {
format,
newOptions,
comment,
- catalogContext);
+ catalogContext,
+ partitionManager);
}
@Override
@@ -302,6 +355,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