[spark] Support partition DDL for catalog-managed Format Table partitions

docs/docs/spark/sql-ddl.md

@@ -206,6 +206,55 @@ CREATE TABLE my_table (
 );
 ```
 
+### Manage Format Table Partitions
+
+For a Format Table whose `metastore.partitioned-table` option is `true`, 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
+
+`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.
+
+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 ('metastore.partitioned-table')`.
+
+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 on a Format Table it
+only takes effect in a REST catalog; elsewhere partitions are discovered from the filesystem.
+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.
+
+For a Format Table, `metastore.partitioned-table` 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. 

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java

@@ -0,0 +1,165 @@
+/*
+ * 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.spark.format;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.format.FormatTablePartitionManager;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.PartitionPathUtils;
+import org.apache.paimon.utils.Preconditions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Repairs a Format Table with catalog-managed partitions by reconciling its partition directories
+ * with its catalog registration, backing {@code MSCK REPAIR TABLE [{ADD|DROP|SYNC} PARTITIONS]}.
+ * Only partition metadata is changed, data files are never touched.
+ */
+public class FormatTablePartitionRepair {
+
+    private FormatTablePartitionRepair() {}
+
+    /**
+     * Repair the partition metadata of a Format Table with catalog-managed partitions. The flag
+     * pair follows Spark's {@code RepairTable} plan (plain MSCK means ADD). Returns the number of
+     * applied actions.
+     */
+    public static int repair(
+            PaimonFormatTable sparkTable, boolean addPartitions, boolean dropPartitions) {
+        Preconditions.checkArgument(
+                addPartitions || dropPartitions,
+                "MSCK REPAIR TABLE must enable ADD and/or DROP partitions");
+        FormatTable formatTable = sparkTable.table();
+        FormatTablePartitionManager partitionManager = formatTable.partitionManager();
+        Preconditions.checkArgument(
+                partitionManager != null,
+                "%s does not have catalog-managed partitions",
+                formatTable.fullName());
+
+        return apply(
+                partitionManager,
+                listFilesystemPartitionSpecs(formatTable),
+                formatTable.partitionKeys(),
+                addPartitions,
+                dropPartitions);
+    }
+
+    private static List<Map<String, String>> listFilesystemPartitionSpecs(FormatTable formatTable) {
+        // Discover partitions from the raw directory names rather than through the table scan:
+        // the scan casts each value to its column type and back (e.g. month=01 -> 1), producing
+        // specs that can no longer round-trip to the real directory. The write path registers the
+        // raw directory value, so a repair must diff against the same raw values to avoid
+        // spuriously adding/dropping partition metadata.
+        boolean onlyValueInPath =
+                new CoreOptions(formatTable.options()).formatTablePartitionOnlyValueInPath();
+        List<Pair<LinkedHashMap<String, String>, Path>> found =
+                PartitionPathUtils.searchPartSpecAndPaths(
+                        formatTable.fileIO(),
+                        new Path(formatTable.location()),
+                        formatTable.partitionKeys().size(),
+                        formatTable.partitionKeys(),
+                        onlyValueInPath);
+        List<Map<String, String>> specs = new ArrayList<>(found.size());
+        for (Pair<LinkedHashMap<String, String>, Path> pair : found) {
+            PartitionPathUtils.validatePartitionSpecForPath(pair.getKey(), onlyValueInPath);
+            specs.add(pair.getKey());
+        }
+        return specs;
+    }
+
+    /**
+     * Diff the filesystem partition set against the catalog registration set and apply the
+     * requested actions. ADD registers "directory exists but unregistered"; DROP is metadata-only
+     * cleanup of "registered but directory missing". Scan-completeness guard: the filesystem
+     * listing that feeds {@code filesystemPartitions} ({@link
+     * PartitionPathUtils#searchPartSpecAndPaths}) fails on any mid-scan LIST error instead of
+     * returning a truncated set, so a DROP diff can only be produced from a complete listing and a
+     * transient failure never deregisters partitions that still exist.
+     */
+    static int apply(
+            FormatTablePartitionManager partitionManager,
+            List<Map<String, String>> filesystemPartitions,
+            List<String> partitionKeys,
+            boolean addPartitions,
+            boolean dropPartitions) {
+        Set<Map<String, String>> registeredPartitions = new HashSet<>();
+        for (Partition partition : partitionManager.listPartitions(Collections.emptyMap())) {
+            registeredPartitions.add(partition.spec());
+        }
+
+        Set<Map<String, String>> filesystemSet = new HashSet<>(filesystemPartitions);
+
+        List<Map<String, String>> addDiff = new ArrayList<>();
+        if (addPartitions) {
+            for (Map<String, String> partition : filesystemPartitions) {
+                if (!registeredPartitions.contains(partition)) {
+                    addDiff.add(partition);
+                }
+            }
+            sortByCanonicalPath(addDiff, partitionKeys);
+        }
+        List<Map<String, String>> dropDiff = new ArrayList<>();
+        if (dropPartitions) {
+            for (Map<String, String> partition : registeredPartitions) {
+                if (!filesystemSet.contains(partition)) {
+                    dropDiff.add(partition);
+                }
+            }
+            sortByCanonicalPath(dropDiff, partitionKeys);
+        }
+
+        // A first repair of a pre-existing table can discover far more partitions than any regular
+        // write. Splitting such a diff into per-request batches is the partition catalog's job;
+        // registration is an idempotent upsert and unregistration ignores missing partitions, so a
+        // mid-way failure leaves a state a rerun converges from.
+        if (!addDiff.isEmpty()) {
+            partitionManager.createPartitions(addDiff, true);
+        }
+        if (!dropDiff.isEmpty()) {
+            partitionManager.dropPartitions(dropDiff);
+        }
+        return addDiff.size() + dropDiff.size();
+    }
+
+    /** Sort partitions by their canonical path for a stable, deterministic apply order. */
+    private static void sortByCanonicalPath(
+            List<Map<String, String>> partitions, List<String> partitionKeys) {
+        partitions.sort(Comparator.comparing(partition -> canonicalPath(partition, partitionKeys)));
+    }
+
+    private static String canonicalPath(Map<String, String> partition, List<String> partitionKeys) {
+        LinkedHashMap<String, String> orderedSpec = new LinkedHashMap<>();
+        for (String key : partitionKeys) {
+            if (partition.containsKey(key)) {
+                orderedSpec.put(key, partition.get(key));
+            }
+        }
+        return PartitionPathUtils.generatePartitionPath(orderedSpec);
+    }
+}

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala

@@ -46,49 +46,43 @@ trait PaimonPartitionManagement extends SupportsAtomicPartitionManagement with L
 
   private def toPaimonPartitions(rows: Array[InternalRow]): Array[java.util.Map[String, String]] = {
     table match {
-      case fileStoreTable: FileStoreTable =>
+      case _: FileStoreTable =>
         val partitionKeys = table.partitionKeys().asScala.toSeq
-        val partitionDefaultName = fileStoreTable.coreOptions().partitionDefaultName()
-        val legacyPartitionName = CoreOptions.fromMap(table.options()).legacyPartitionName
-
         rows.map {
           r =>
-            val partitionFieldCount = r.numFields
             require(
-              partitionFieldCount <= partitionKeys.length,
-              s"Partition values length $partitionFieldCount exceeds partition keys " +
+              r.numFields <= partitionKeys.length,
+              s"Partition values length ${r.numFields} exceeds partition keys " +
                 s"${partitionKeys.mkString("[", ", ", "]")}."
             )
-            val partitionNames = partitionKeys.take(partitionFieldCount)
-            val currentPartitionRowType =
-              if (partitionFieldCount == partitionRowType.getFieldCount) {
-                partitionRowType
-              } else {
-                TypeUtils.project(table.rowType, partitionNames.asJava)
-              }
-            val currentPartitionSchema =
-              if (partitionFieldCount == partitionSchema.length) {
-                partitionSchema
-              } else {
-                SparkTypeUtils.fromPaimonRowType(currentPartitionRowType)
-              }
-            val rowConverter = CatalystTypeConverters.createToScalaConverter(
-              CharVarcharUtils.replaceCharVarcharWithString(currentPartitionSchema))
-            val rowDataPartitionComputer = new InternalRowPartitionComputer(
-              partitionDefaultName,
-              currentPartitionRowType,
-              partitionNames.toArray,
-              legacyPartitionName
-            )
-
-            rowDataPartitionComputer.generatePartValues(
-              new SparkRow(currentPartitionRowType, rowConverter(r).asInstanceOf[Row]))
+            toPaimonPartition(r, partitionKeys.take(r.numFields))
         }
       case _ =>
         throw new UnsupportedOperationException("Only FileStoreTable supports partitions.")
     }
   }
 
+  protected def toPaimonPartition(
+      row: InternalRow,
+      partitionNames: Seq[String]): java.util.Map[String, String] = {
+    val coreOptions = CoreOptions.fromMap(table.options())
+    val partitionDefaultName = coreOptions.partitionDefaultName()
+    val legacyPartitionName = coreOptions.legacyPartitionName
+    val currentPartitionRowType = TypeUtils.project(table.rowType, partitionNames.asJava)
+    val currentPartitionSchema = SparkTypeUtils.fromPaimonRowType(currentPartitionRowType)
+    val rowConverter = CatalystTypeConverters.createToScalaConverter(
+      CharVarcharUtils.replaceCharVarcharWithString(currentPartitionSchema))
+    val rowDataPartitionComputer = new InternalRowPartitionComputer(
+      partitionDefaultName,
+      currentPartitionRowType,
+      partitionNames.toArray,
+      legacyPartitionName
+    )
+
+    rowDataPartitionComputer.generatePartValues(
+      new SparkRow(currentPartitionRowType, rowConverter(row).asInstanceOf[Row]))
+  }
+
   override def dropPartitions(rows: Array[InternalRow]): Boolean = {
     table match {
       case fileStoreTable: FileStoreTable =>