[spark] Add managed Format Table partition administration

docs/docs/flink/sql-ddl.md

@@ -34,6 +34,22 @@ Paimon catalogs currently support three types of metastores:
 
 See [CatalogOptions](../maintenance/configurations#catalogoptions) for detailed options when creating a catalog.
 
+:::info
+
+For Format Tables, `metastore.partitioned-table = true` enables catalog-managed 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` (catalog-level `table-default.*` options participate in the
+effective options); other catalogs keep their previous behavior and treat the option as inert.
+
+A Format Table that carries `metastore.partitioned-table = true` where it cannot be honored
+(e.g. in a Hive catalog, or on an external table) still loads — the option is ignored, the table
+behaves as an unmanaged Format Table, and a warning is logged. To remove the option permanently,
+use `ALTER TABLE my_table RESET ('metastore.partitioned-table')` where the catalog supports
+altering the table.
+
+:::
+
 ### 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/procedures.md

@@ -586,5 +586,37 @@ This section introduce all available spark procedures about paimon.
          CALL sys.rescale(table => 'default.T', bucket_num => 16, partitions => 'dt=20250217,hh=08;dt=20250217,hh=09')<br/>
       </td>
    </tr>
+   <tr>
+      <td>list_format_table_partitions</td>
+      <td>
+         To list partitions of a managed Format Table (a Format Table with 'metastore.partitioned-table' = 'true') page by page. Each output row contains a partition path and the next_page_token to pass into the next call; the token is null on the last page. When a where predicate is given, each page returns the matching rows of one raw catalog page, so filtered pages can carry fewer than limit rows; keep calling with the returned token until it is null. Arguments:
+            <li>table: the target managed Format Table identifier. Cannot be empty.</li>
+            <li>where: partition predicate. Left empty for all partitions.</li>
+            <li>limit: the maximum number of partitions to return in one page, between 1 and 10000. Default is 1000.</li>
+            <li>page_token: the next_page_token returned by the previous call. Left empty for the first page.</li>
+      </td>
+      <td>
+         CALL sys.list_format_table_partitions(table => 'default.T')<br/><br/>
+         CALL sys.list_format_table_partitions(table => 'default.T', where => 'dt >= 20250101', limit => 500)<br/><br/>
+         CALL sys.list_format_table_partitions(table => 'default.T', page_token => 'token-from-previous-call')<br/>
+      </td>
+   </tr>
+   <tr>
+      <td>sync_format_table_metadata</td>
+      <td>
+         To reconcile the partition metadata of a managed Format Table (a Format Table with 'metastore.partitioned-table' = 'true') with the partition directories on the filesystem. Only partition metadata is changed, data files are never touched. Defaults to a dry run which only reports the planned actions; pass dry_run => false to apply them. 'MSCK REPAIR TABLE' is the SQL equivalent which always applies the changes. Arguments:
+            <li>table: the target managed Format Table identifier. Cannot be empty.</li>
+            <li>mode: 'ADD' registers filesystem partitions missing from the catalog, 'DROP' unregisters catalog partitions whose directories no longer exist, 'SYNC' does both. Default is 'ADD'.</li>
+            <li>dry_run: only report the planned actions without applying them. Default is true.</li>
+      </td>
+      <td>
+         -- report partitions that would be registered<br/>
+         CALL sys.sync_format_table_metadata(table => 'default.T')<br/><br/>
+         -- register missing partitions<br/>
+         CALL sys.sync_format_table_metadata(table => 'default.T', dry_run => false)<br/><br/>
+         -- fully synchronize directories and metadata<br/>
+         CALL sys.sync_format_table_metadata(table => 'default.T', mode => 'SYNC', dry_run => false)<br/>
+      </td>
+   </tr>
    </tbody>
 </table>

docs/docs/spark/sql-ddl.md

@@ -206,6 +206,52 @@ CREATE TABLE my_table (
 );
 ```
 
+### Manage Format Table Partitions
+
+For managed Format Tables, whose `metastore.partitioned-table` option is `true`, partition
+metadata is stored in the catalog 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 that is not managed, `ADD PARTITION`, `DROP PARTITION` and
+`MSCK REPAIR TABLE` fail with an error.
+
+`SHOW PARTITIONS` throws an error when the number of partitions exceeds
+`spark.paimon.format-table.show-partitions.max-results` (default: 10000). For larger tables, use
+`CALL sys.list_format_table_partitions` to list partitions page by page.
+
+`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
+
+`metastore.partitioned-table = true` 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. Other catalogs treat the option as inert.
+
+A table that carries `metastore.partitioned-table = true` where it cannot be honored (a non-REST
+catalog, or an external table) loads as an unmanaged Format Table and a warning is logged. To
+remove the option, use `ALTER TABLE my_table UNSET TBLPROPERTIES ('metastore.partitioned-table')`.
+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` or
+`CALL sys.sync_format_table_metadata`.
+
+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 managed reads, or
+run `MSCK REPAIR TABLE` / `CALL sys.sync_format_table_metadata` afterwards, since data written by
+an older writer is not visible to a managed read until its partitions are registered.
+
+:::
+
 ### 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

@@ -1029,7 +1029,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 format tables in a REST catalog, setting this option makes partition metadata catalog-managed: partitions must be registered in the catalog, and partition listing does not fall back to filesystem discovery.</td>
         </tr>
         <tr>
             <td><h5>metastore.tag-to-partition</h5></td>

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

@@ -1823,7 +1823,10 @@ 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 format tables in a REST catalog, setting this option makes partition "
+                                    + "metadata catalog-managed: partitions must be registered in the catalog, "
+                                    + "and partition listing does not fall back to filesystem discovery.");
 
     public static final ConfigOption<String> METASTORE_TAG_TO_PARTITION =
             key("metastore.tag-to-partition")

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/Catalog.java

@@ -670,6 +670,17 @@ default boolean supportsListTableByType() {
         return false;
     }
 
+    /**
+     * Whether this catalog can be the source of truth for partitions of managed Format Tables
+     * ({@code type=format-table} with {@code metastore.partitioned-table=true}), i.e. whether it
+     * implements the partition APIs that managed Format Table scans and commits rely on. Catalogs
+     * without this capability load such tables with the option ignored, falling back to filesystem
+     * partition discovery.
+     */
+    default boolean supportsManagedFormatTablePartitions() {
+        return false;
+    }
+
     // ==================== Version management methods ==========================
 
     /**

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.FormatTableCatalogProvider;
 import org.apache.paimon.table.iceberg.IcebergTable;
 import org.apache.paimon.table.lance.LanceTable;
 import org.apache.paimon.table.object.ObjectTable;
@@ -54,6 +55,9 @@
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.Preconditions;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
@@ -85,6 +89,8 @@
 /** Utils for {@link Catalog}. */
 public class CatalogUtils {
 
+    private static final Logger LOG = LoggerFactory.getLogger(CatalogUtils.class);
+
     public static Path path(String warehouse, String database, String table) {
         return new Path(String.format("%s/%s.db/%s", warehouse, database, table));
     }
@@ -173,6 +179,7 @@ private static void validateFormatTableOptions(Options options, boolean dataToke
                 options.get(PRIMARY_KEY) == null,
                 "Cannot define %s for format table.",
                 PRIMARY_KEY.key());
+        validateManagedFormatTableOptions(options);
         if (dataTokenEnabled && options.get(PATH) == null) {
             checkArgument(
                     options.get(FORMAT_TABLE_IMPLEMENTATION)
@@ -213,6 +220,44 @@ private static void validateFormatTableOptions(Options options, boolean dataToke
         }
     }
 
+    /** Validate options which are specific to catalog-managed format tables. */
+    public static void validateManagedFormatTableOptions(Map<String, String> tableOptions) {
+        validateManagedFormatTableOptions(Options.fromMap(tableOptions));
+    }
+
+    private static void validateManagedFormatTableOptions(Options options) {
+        if (options.get(CoreOptions.METASTORE_PARTITIONED_TABLE)) {
+            checkArgument(
+                    options.get(FORMAT_TABLE_IMPLEMENTATION)
+                            != CoreOptions.FormatTableImplementation.ENGINE,
+                    "Cannot define %s=true together with %s=engine for a managed format table.",
+                    CoreOptions.METASTORE_PARTITIONED_TABLE.key(),
+                    FORMAT_TABLE_IMPLEMENTATION.key());
+        }
+    }
+
+    /** Validate that a catalog can persist the requested managed Format Table. */
+    public static void validateManagedFormatTableCatalog(
+            Identifier identifier,
+            Map<String, String> tableOptions,
+            Catalog catalog,
+            boolean isExternal) {
+        CoreOptions options = CoreOptions.fromMap(tableOptions);
+        if (options.type() != TableType.FORMAT_TABLE || !options.partitionedTableInMetastore()) {
+            return;
+        }
+        checkArgument(
+                !isExternal,
+                "Managed format table %s must be an internal table.",
+                identifier.getFullName());
+        checkArgument(
+                catalog.supportsManagedFormatTablePartitions(),
+                "Managed format table %s requires a catalog with catalog-managed partition support"
+                        + " (e.g. a REST catalog), but %s does not support it.",
+                identifier.getFullName(),
+                catalog.getClass().getSimpleName());
+    }
+
     public static void validateNamePattern(Catalog catalog, String namePattern) {
         if (Objects.nonNull(namePattern) && !catalog.supportsListByPattern()) {
             throw new UnsupportedOperationException(
@@ -271,6 +316,34 @@ public static List<Partition> listPartitionsFromFileSystem(
         return partitions;
     }
 
+    /**
+     * Why the managed Format Table semantics cannot be honored when loading the table from this
+     * catalog, or {@code null} if they can.
+     */
+    @Nullable
+    private static String managedFormatTableLoadProblem(
+            Catalog catalog, TableMetadata metadata, Map<String, String> tableOptions) {
+        if (!catalog.supportsManagedFormatTablePartitions()) {
+            return String.format(
+                    "catalog %s does not support catalog-managed Format Table partitions",
+                    catalog.getClass().getSimpleName());
+        }
+        if (metadata.isExternal()) {
+            return "managed format tables must be internal tables";
+        }
+        if (Options.fromMap(tableOptions).get(FORMAT_TABLE_IMPLEMENTATION)
+                == CoreOptions.FormatTableImplementation.ENGINE) {
+            return String.format(
+                    "the option conflicts with %s=engine", FORMAT_TABLE_IMPLEMENTATION.key());
+        }
+        if (catalog.catalogLoader() == null) {
+            return String.format(
+                    "catalog %s does not provide a catalog loader",
+                    catalog.getClass().getSimpleName());
+        }
+        return null;
+    }
+
     /**
      * Load table from {@link Catalog}, this table can be:
      *
@@ -304,7 +377,30 @@ public static Table loadTable(
         Function<Path, FileIO> dataFileIO = metadata.isExternal() ? externalFileIO : internalFileIO;
 
         if (options.type() == TableType.FORMAT_TABLE) {
-            return toFormatTable(identifier, schema, dataFileIO, catalogContext);
+            FormatTableCatalogProvider catalogProvider = null;
+            if (options.partitionedTableInMetastore()) {
+                String problem = managedFormatTableLoadProblem(catalog, metadata, schema.options());
+                if (problem == null) {
+                    catalogProvider =
+                            new FormatTableCatalogProvider(identifier, catalog.catalogLoader());
+                } else {
+                    // Existing tables may carry the option from a version (or catalog) where it
+                    // was inert; refusing to load them would leave no SQL way to repair the
+                    // table. Keep them readable with filesystem partition discovery instead.
+                    LOG.warn(
+                            "Ignoring '{}=true' on format table {}: {}. Falling back to "
+                                    + "filesystem partition discovery, the table behaves as an "
+                                    + "unmanaged format table. Use ALTER TABLE ... RESET/UNSET "
+                                    + "to remove the option.",
+                            CoreOptions.METASTORE_PARTITIONED_TABLE.key(),
+                            identifier,
+                            problem);
+                    Map<String, String> effectiveOptions = new HashMap<>(schema.options());
+                    effectiveOptions.remove(CoreOptions.METASTORE_PARTITIONED_TABLE.key());
+                    schema = schema.copy(effectiveOptions);
+                }
+            }
+            return toFormatTable(identifier, schema, dataFileIO, catalogContext, catalogProvider);
         }
 
         if (options.type() == TableType.OBJECT_TABLE) {
@@ -453,7 +549,8 @@ private static FormatTable toFormatTable(
             Identifier identifier,
             TableSchema schema,
             Function<Path, FileIO> fileIO,
-            CatalogContext catalogContext) {
+            CatalogContext catalogContext,
+            @Nullable FormatTableCatalogProvider catalogProvider) {
         Map<String, String> options = schema.options();
         FormatTable.Format format =
                 FormatTable.parseFormat(
@@ -471,6 +568,7 @@ private static FormatTable toFormatTable(
                 .options(options)
                 .comment(schema.comment())
                 .catalogContext(catalogContext)
+                .catalogProvider(catalogProvider)
                 .build();
     }
 

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

@@ -198,6 +198,11 @@ public boolean supportsListTableByType() {
         return wrapped.supportsListTableByType();
     }
 
+    @Override
+    public boolean supportsManagedFormatTablePartitions() {
+        return wrapped.supportsManagedFormatTablePartitions();
+    }
+
     @Override
     public boolean supportsVersionManagement() {
         return wrapped.supportsVersionManagement();