[core][rest] Support catalog-managed partitions for Format Table reads and writes

docs/docs/flink/sql-ddl.md

@@ -34,6 +34,29 @@ Paimon catalogs currently support three types of metastores:
 
 See [CatalogOptions](../maintenance/configurations#catalogoptions) for detailed options when creating a catalog.
 
+:::info
+
+For an internal Format Table in a REST catalog, `metastore.partitioned-table = true` makes the
+catalog the source of truth for partitions, which requires an internal table in a catalog that supports it (currently the REST
+catalog) and cannot be combined with `format-table.implementation = engine`. On a Paimon table
+the same option keeps its existing meaning (synchronize partitions into the metastore); on a
+Format Table it only takes effect in a REST catalog, and elsewhere partitions are still discovered
+from the filesystem.
+
+**A Flink job reads only the partitions the catalog knows.** Directories written before the option
+was enabled, by an older writer, or by anything that does not register what it wrote are invisible,
+and a table whose catalog holds no partitions reads as empty. Flink has no SQL command to register
+them: use Spark's `MSCK REPAIR TABLE` or the catalog's partition API. Flink writes on a current
+version do register the partitions they produce.
+
+In a REST catalog, asking for catalog-managed partitions on a table that cannot have them — an
+external table, or `format-table.implementation = engine` — fails. In any other catalog the option
+keeps the meaning it has always had on a Format Table — none — and partitions come from the
+filesystem. Removing the option needs a catalog that can alter the table; Flink's Format Table path
+cannot, so use another engine.
+
+:::
+
 ### Create Filesystem Catalog
 
 The following Flink SQL registers and uses a Paimon catalog named `my_catalog`. Metadata and table files are stored under `hdfs:///path/to/warehouse`.

docs/generated/core_configuration.html

@@ -1041,7 +1041,8 @@
             <td>Boolean</td>
             <td>Whether to create this table as a partitioned table in metastore.
 For example, if you want to list all partitions of a Paimon table in Hive, you need to create this table as a partitioned table in Hive metastore.
-This config option does not affect the default filesystem metastore.</td>
+This config option does not affect the default filesystem metastore.
+For an internal format table in a REST catalog, it also makes the catalog own the table's partitions: a scan reads the partitions registered there and a write registers the ones it wrote, instead of listing the table directory.</td>
         </tr>
         <tr>
             <td><h5>metastore.tag-to-partition</h5></td>

docs/static/rest-catalog-open-api.yaml

@@ -1030,7 +1030,7 @@ paths:
       tags:
         - partition
       summary: Drop partitions
-      description: Unregisters partitions from the catalog. The server never deletes data files; managed format table data deletion is performed by the client afterwards.
+      description: Unregisters partitions from the catalog. The server never deletes data files; for a format table with catalog-managed partitions, data deletion is performed by the client afterwards.
       operationId: dropPartitions
       parameters:
         - name: prefix

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -1850,7 +1850,11 @@ public String toString() {
                             "Whether to create this table as a partitioned table in metastore.\n"
                                     + "For example, if you want to list all partitions of a Paimon table in Hive, "
                                     + "you need to create this table as a partitioned table in Hive metastore.\n"
-                                    + "This config option does not affect the default filesystem metastore.");
+                                    + "This config option does not affect the default filesystem metastore.\n"
+                                    + "For an internal format table in a REST catalog, it also makes the "
+                                    + "catalog own the table's partitions: a scan reads the partitions "
+                                    + "registered there and a write registers the ones it wrote, instead of "
+                                    + "listing the table directory.");
 
     public static final ConfigOption<String> METASTORE_TAG_TO_PARTITION =
             key("metastore.tag-to-partition")

paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java

@@ -39,6 +39,7 @@
 import org.apache.paimon.table.FormatTable;
 import org.apache.paimon.table.Table;
 import org.apache.paimon.table.TableSnapshot;
+import org.apache.paimon.table.format.FormatTablePartitionManager;
 import org.apache.paimon.table.iceberg.IcebergTable;
 import org.apache.paimon.table.lance.LanceTable;
 import org.apache.paimon.table.object.ObjectTable;
@@ -213,6 +214,41 @@ private static void validateFormatTableOptions(Options options, boolean dataToke
         }
     }
 
+    /** Validate options which are specific to a Format Table with catalog-managed partitions. */
+    public static void validateCatalogManagedPartitionOptions(Map<String, String> tableOptions) {
+        validateCatalogManagedPartitionOptions(Options.fromMap(tableOptions));
+    }
+
+    private static void validateCatalogManagedPartitionOptions(Options options) {
+        // The caller has already established that this is a format table with catalog-managed
+        // partitions; only the engine implementation is incompatible with them.
+        checkArgument(
+                options.get(FORMAT_TABLE_IMPLEMENTATION)
+                        != CoreOptions.FormatTableImplementation.ENGINE,
+                "Cannot combine catalog-managed partitions (%s=true) with %s=engine: the engine "
+                        + "implementation reads the table directory itself.",
+                CoreOptions.METASTORE_PARTITIONED_TABLE.key(),
+                FORMAT_TABLE_IMPLEMENTATION.key());
+    }
+
+    /**
+     * Validate a create or replace request that asks for catalog-managed partitions on a Format
+     * Table: the option combination must be valid and the table must be internal. Only the REST
+     * catalog calls this; other catalogs keep treating the option as inert.
+     */
+    public static void validateCatalogManagedFormatTablePartitions(
+            Identifier identifier, Map<String, String> tableOptions, boolean isExternal) {
+        CoreOptions options = CoreOptions.fromMap(tableOptions);
+        if (options.type() != TableType.FORMAT_TABLE || !options.partitionedTableInMetastore()) {
+            return;
+        }
+        validateCatalogManagedPartitionOptions(Options.fromMap(tableOptions));
+        checkArgument(
+                !isExternal,
+                "Catalog-managed partitions are only supported for internal tables, but format table %s is external.",
+                identifier.getFullName());
+    }
+
     public static void validateNamePattern(Catalog catalog, String namePattern) {
         if (Objects.nonNull(namePattern) && !catalog.supportsListByPattern()) {
             throw new UnsupportedOperationException(
@@ -304,7 +340,21 @@ public static Table loadTable(
         Function<Path, FileIO> dataFileIO = metadata.isExternal() ? externalFileIO : internalFileIO;
 
         if (options.type() == TableType.FORMAT_TABLE) {
-            return toFormatTable(identifier, schema, dataFileIO, catalogContext);
+            FormatTablePartitionManager partitionManager = null;
+            if (options.partitionedTableInMetastore()) {
+                checkArgument(
+                        isRestCatalog,
+                        "Format table %s asks for catalog-managed partitions with %s=true, which "
+                                + "is only available in a REST catalog.",
+                        identifier.getFullName(),
+                        CoreOptions.METASTORE_PARTITIONED_TABLE.key());
+                validateCatalogManagedFormatTablePartitions(
+                        identifier, schema.options(), metadata.isExternal());
+                partitionManager =
+                        FormatTablePartitionManager.create(
+                                identifier, schema.partitionKeys(), catalog.catalogLoader());
+            }
+            return toFormatTable(identifier, schema, dataFileIO, catalogContext, partitionManager);
         }
 
         if (options.type() == TableType.OBJECT_TABLE) {
@@ -453,7 +503,8 @@ private static FormatTable toFormatTable(
             Identifier identifier,
             TableSchema schema,
             Function<Path, FileIO> fileIO,
-            CatalogContext catalogContext) {
+            CatalogContext catalogContext,
+            @Nullable FormatTablePartitionManager partitionManager) {
         Map<String, String> options = schema.options();
         FormatTable.Format format =
                 FormatTable.parseFormat(
@@ -471,6 +522,7 @@ private static FormatTable toFormatTable(
                 .options(options)
                 .comment(schema.comment())
                 .catalogContext(catalogContext)
+                .partitionManager(partitionManager)
                 .build();
     }
 

paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java

@@ -94,6 +94,8 @@
 import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemTable;
 import static org.apache.paimon.catalog.CatalogUtils.isSystemDatabase;
 import static org.apache.paimon.catalog.CatalogUtils.listPartitionsFromFileSystem;
+import static org.apache.paimon.catalog.CatalogUtils.validateCatalogManagedFormatTablePartitions;
+import static org.apache.paimon.catalog.CatalogUtils.validateCatalogManagedPartitionOptions;
 import static org.apache.paimon.catalog.CatalogUtils.validateCreateTable;
 import static org.apache.paimon.options.CatalogOptions.CASE_SENSITIVE;
 
@@ -567,8 +569,12 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx
             checkNotBranch(identifier, "createTable");
             checkNotSystemTable(identifier, "createTable");
             validateCreateTable(schema, dataTokenEnabled);
-            createExternalTablePathIfNotExist(schema);
             tableDefaultOptions.forEach(schema.options()::putIfAbsent);
+            // Defaults participate in the catalog-managed partition combination, so validate
+            // the effective options rather than only the explicit ones.
+            validateCatalogManagedFormatTablePartitions(
+                    identifier, schema.options(), schema.options().containsKey(PATH.key()));
+            createExternalTablePathIfNotExist(schema);
             Schema newSchema = inferSchemaIfExternalPaimonTable(schema);
             api.createTable(identifier, newSchema);
         } catch (AlreadyExistsException e) {
@@ -645,8 +651,13 @@ public void replaceTable(Identifier identifier, Schema newSchema, boolean ignore
         checkNotBranch(identifier, "replaceTable");
         checkNotSystemTable(identifier, "replaceTable");
         validateCreateTable(newSchema, dataTokenEnabled);
+        tableDefaultOptions.forEach(newSchema.options()::putIfAbsent);
+        // Defaults participate in the catalog-managed partition combination, so validate the
+        // effective options rather than only the explicit ones. Externality is not validated
+        // client-side here: a round-tripped schema of an internal table may carry the synthetic
+        // path option, and the server remains the authority for replace semantics.
+        validateCatalogManagedPartitionOptions(newSchema.options());
         try {
-            tableDefaultOptions.forEach(newSchema.options()::putIfAbsent);
             api.replaceTable(identifier, newSchema);
         } catch (NoSuchResourceException e) {
             if (!ignoreIfNotExists) {
@@ -758,9 +769,9 @@ public void createPartitions(
     public void dropPartitions(Identifier identifier, List<Map<String, String>> partitions)
             throws TableNotExistException {
         Table table = getTable(identifier);
-        if (isCatalogManagedFormatTable(table)) {
-            // Managed format table: metadata-only unregistration on the server. Data deletion is
-            // the caller's responsibility (performed with the table FileIO afterwards).
+        if (hasCatalogManagedPartitions(table)) {
+            // Unregistering is metadata-only on the server; deleting the data is the caller's
+            // job, done with the table FileIO afterwards.
             try {
                 api.dropPartitions(identifier, partitions, true);
             } catch (NoSuchResourceException e) {
@@ -772,7 +783,7 @@ public void dropPartitions(Identifier identifier, List<Map<String, String>> part
             }
             return;
         }
-        // Non-managed tables keep the default truncate-based data semantics.
+        // Every other table keeps the default truncate-based data semantics.
         try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) {
             commit.truncatePartitions(partitions);
         } catch (Exception e) {
@@ -790,7 +801,7 @@ public List<Partition> listPartitions(Identifier identifier) throws TableNotExis
             throw new TableNoPermissionException(identifier, e);
         } catch (NotImplementedException e) {
             Table table = getTable(identifier);
-            if (isCatalogManagedFormatTable(table)) {
+            if (hasCatalogManagedPartitions(table)) {
                 throw e;
             }
             return listPartitionsFromFileSystem(table);
@@ -812,7 +823,7 @@ public PagedList<Partition> listPartitionsPaged(
             throw new TableNoPermissionException(identifier, e);
         } catch (NotImplementedException e) {
             Table table = getTable(identifier);
-            if (isCatalogManagedFormatTable(table)) {
+            if (hasCatalogManagedPartitions(table)) {
                 throw e;
             }
             return new PagedList<>(listPartitionsFromFileSystem(table), null);
@@ -831,7 +842,7 @@ public List<Partition> 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) {