[core][spark] Support managed Format Table partition operations

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 Format Tables, `format-table.partition-source = rest` 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`. The REST catalog
+validates this combination on `CREATE TABLE`, with catalog-level `table-default.*` options
+participating in the effective options. Other catalogs keep discovering partitions from the
+filesystem and ignore the option.
+
+**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/docs/spark/sql-ddl.md

@@ -206,6 +206,55 @@ CREATE TABLE my_table (
 );
 ```
 
+### Manage Format Table Partitions
+
+For a Format Table whose `format-table.partition-source` option is `rest`, the catalog holds the
+partitions and Spark supports the standard partition DDL:
+
+```sql
+ALTER TABLE my_table ADD PARTITION (dt='2025-01-01');
+ALTER TABLE my_table DROP PARTITION (dt='2025-01-01');
+MSCK REPAIR TABLE my_table;
+SHOW PARTITIONS my_table;
+```
+
+On a Format Table whose partitions are discovered from the filesystem, `ADD PARTITION`,
+`DROP PARTITION` and `MSCK REPAIR TABLE` fail with an error.
+
+`ADD PARTITION` creates the partition directory and registers the partition; querying a newly
+added partition before any data is written returns no rows. `DROP PARTITION` unregisters the
+partition and deletes its directory.
+
+:::info
+
+`format-table.partition-source = rest` enables catalog-managed partitions, which requires an
+internal Format Table in a catalog that supports it (currently the REST catalog) and cannot be
+combined with `format-table.implementation = engine`. The REST catalog validates this
+combination on `CREATE TABLE`, with catalog-level table defaults
+(`spark.sql.catalog.paimon.table-default.*`) participating in the effective options: a default
+that makes the combination invalid fails the DDL.
+
+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, rather than handing back a
+table whose options say one thing and whose partitions come from somewhere else. Remove the option
+with `ALTER TABLE my_table UNSET TBLPROPERTIES ('format-table.partition-source')`.
+
+In any other catalog the option keeps the meaning it has always had on a Format Table — none. Such
+a table loads and reads its partitions from the filesystem, so it is a format-table option, so nothing a
+Paimon table sets can turn it on by accident.
+On a REST catalog, an existing Format Table whose partitions were never registered reads as empty
+until you register them with `MSCK REPAIR TABLE my_table`.
+
+Mixed-version note: only writers that support catalog-managed partitions register the partitions
+they produce. During a rolling upgrade, upgrade all writers before relying on catalog-managed
+partitions, or run `MSCK REPAIR TABLE` afterwards, since data written by an older writer is not
+visible until its partitions are registered.
+
+Setting `format-table.partition-source` only changes where partitions come from; it does not
+change the table's managed/external ownership.
+
+:::
+
 ### Create External Table
 
 When the catalog's `metastore` type is `hive`, if the `location` is specified when creating a table, that table will be considered an external table; otherwise, it will be a managed table. 

docs/generated/core_configuration.html

@@ -734,6 +734,12 @@
             <td><p>Enum</p></td>
             <td>Format table uses paimon or engine.<br /><br />Possible values:<ul><li>"paimon": Paimon format table implementation.</li><li>"engine": Engine format table implementation.</li></ul></td>
         </tr>
+        <tr>
+            <td><h5>format-table.partition-source</h5></td>
+            <td style="word-wrap: break-word;">filesystem</td>
+            <td><p>Enum</p></td>
+            <td>Where the partitions of a format table come from. With 'filesystem' they are found by listing the table directory. With 'rest' the catalog holds them: a scan reads the partitions registered there and a write registers the ones it wrote, so a directory nobody registered is not part of the table. 'rest' needs an internal table in a REST catalog and cannot be combined with format-table.implementation=engine.<br /><br />Possible values:<ul><li>"filesystem": Partitions are found by listing the table directory.</li><li>"rest": Partitions are the ones registered in the REST catalog.</li></ul></td>
+        </tr>
         <tr>
             <td><h5>format-table.partition-path-only-value</h5></td>
             <td style="word-wrap: break-word;">false</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

@@ -2520,6 +2520,19 @@ public String toString() {
                     .defaultValue(FormatTableImplementation.PAIMON)
                     .withDescription("Format table uses paimon or engine.");
 
+    public static final ConfigOption<PartitionSource> FORMAT_TABLE_PARTITION_SOURCE =
+            key("format-table.partition-source")
+                    .enumType(PartitionSource.class)
+                    .defaultValue(PartitionSource.FILESYSTEM)
+                    .withDescription(
+                            "Where the partitions of a format table come from. With 'filesystem' "
+                                    + "they are found by listing the table directory. With 'rest' "
+                                    + "the catalog holds them: a scan reads the partitions "
+                                    + "registered there and a write registers the ones it wrote, "
+                                    + "so a directory nobody registered is not part of the table. "
+                                    + "'rest' needs an internal table in a REST catalog and cannot "
+                                    + "be combined with format-table.implementation=engine.");
+
     public static final ConfigOption<Boolean> FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH =
             ConfigOptions.key("format-table.partition-path-only-value")
                     .booleanType()
@@ -4235,6 +4248,10 @@ public boolean formatTableImplementationIsPaimon() {
         return options.get(FORMAT_TABLE_IMPLEMENTATION) == FormatTableImplementation.PAIMON;
     }
 
+    public boolean formatTablePartitionsFromCatalog() {
+        return options.get(FORMAT_TABLE_PARTITION_SOURCE) == PartitionSource.REST;
+    }
+
     public boolean formatTablePartitionOnlyValueInPath() {
         return options.get(FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH);
     }
@@ -5249,6 +5266,31 @@ public enum PartitionSinkStrategy {
         PARTITION_DYNAMIC
     }
 
+    /** Where the partitions of a format table come from. */
+    public enum PartitionSource implements DescribedEnum {
+        FILESYSTEM("filesystem", "Partitions are found by listing the table directory."),
+        REST("rest", "Partitions are the ones registered in the REST catalog.");
+
+        private final String value;
+
+        private final String description;
+
+        PartitionSource(String value, String description) {
+            this.value = value;
+            this.description = description;
+        }
+
+        @Override
+        public String toString() {
+            return value;
+        }
+
+        @Override
+        public InlineElement getDescription() {
+            return text(description);
+        }
+    }
+
     /** Specifies the implementation of format table. */
     public enum FormatTableImplementation implements DescribedEnum {
         PAIMON("paimon", "Paimon format table implementation."),

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

@@ -338,35 +338,45 @@ public List<Partition> listPartitions(Identifier identifier) throws TableNotExis
     @Override
     public void createPartitions(Identifier identifier, List<Map<String, String>> partitions)
             throws TableNotExistException {
-        wrapped.createPartitions(identifier, partitions);
-        if (partitionCache != null) {
-            partitionCache.invalidate(identifier);
+        try {
+            wrapped.createPartitions(identifier, partitions);
+        } finally {
+            invalidatePartitionCache(identifier);
         }
     }
 
     @Override
     public void createPartitions(
             Identifier identifier, List<Map<String, String>> partitions, boolean ignoreIfExists)
             throws TableNotExistException {
-        wrapped.createPartitions(identifier, partitions, ignoreIfExists);
-        if (partitionCache != null) {
-            partitionCache.invalidate(identifier);
+        try {
+            wrapped.createPartitions(identifier, partitions, ignoreIfExists);
+        } finally {
+            invalidatePartitionCache(identifier);
         }
     }
 
     @Override
     public void dropPartitions(Identifier identifier, List<Map<String, String>> partitions)
             throws TableNotExistException {
-        wrapped.dropPartitions(identifier, partitions);
-        if (partitionCache != null) {
-            partitionCache.invalidate(identifier);
+        try {
+            wrapped.dropPartitions(identifier, partitions);
+        } finally {
+            invalidatePartitionCache(identifier);
         }
     }
 
     @Override
     public void alterPartitions(Identifier identifier, List<PartitionStatistics> partitions)
             throws TableNotExistException {
-        wrapped.alterPartitions(identifier, partitions);
+        try {
+            wrapped.alterPartitions(identifier, partitions);
+        } finally {
+            invalidatePartitionCache(identifier);
+        }
+    }
+
+    private void invalidatePartitionCache(Identifier identifier) {
         if (partitionCache != null) {
             partitionCache.invalidate(identifier);
         }

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;
@@ -68,6 +69,7 @@
 
 import static org.apache.paimon.CoreOptions.AUTO_CREATE;
 import static org.apache.paimon.CoreOptions.FORMAT_TABLE_IMPLEMENTATION;
+import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_SOURCE;
 import static org.apache.paimon.CoreOptions.PARTITION_DEFAULT_NAME;
 import static org.apache.paimon.CoreOptions.PARTITION_GENERATE_LEGACY_NAME;
 import static org.apache.paimon.CoreOptions.PATH;
@@ -213,6 +215,41 @@ private static void validateFormatTableOptions(Options options, boolean dataToke
         }
     }
 
+    /** Validate options which are specific to Format Tables with catalog-managed partitions. */
+    public static void validateCatalogManagedPartitionOptions(Map<String, String> tableOptions) {
+        validateCatalogManagedPartitionOptions(Options.fromMap(tableOptions));
+    }
+
+    private static void validateCatalogManagedPartitionOptions(Options options) {
+        if (options.get(FORMAT_TABLE_PARTITION_SOURCE) == CoreOptions.PartitionSource.REST) {
+            checkArgument(
+                    options.get(FORMAT_TABLE_IMPLEMENTATION)
+                            != CoreOptions.FormatTableImplementation.ENGINE,
+                    "Cannot define %s=rest together with %s=engine: the engine implementation reads the table directory itself.",
+                    FORMAT_TABLE_PARTITION_SOURCE.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) {
+        validateCatalogManagedPartitionOptions(tableOptions);
+        CoreOptions options = CoreOptions.fromMap(tableOptions);
+        if (options.type() != TableType.FORMAT_TABLE
+                || !options.formatTablePartitionsFromCatalog()) {
+            return;
+        }
+        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 +341,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.formatTablePartitionsFromCatalog()) {
+                checkArgument(
+                        isRestCatalog,
+                        "Format table %s asks for its partitions with %s=rest, which is only "
+                                + "available in a REST catalog.",
+                        identifier.getFullName(),
+                        FORMAT_TABLE_PARTITION_SOURCE.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 +504,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 +523,7 @@ private static FormatTable toFormatTable(
                 .options(options)
                 .comment(schema.comment())
                 .catalogContext(catalogContext)
+                .partitionManager(partitionManager)
                 .build();
     }