[core] Support success-file partition mark done for Format Table

docs/docs/flink/procedures/table-operations.md

@@ -379,6 +379,9 @@ Mark partitions as done. Arguments:
 
 - `partitions`: semicolon-separated partition specs.
 
+A Format Table only supports the `success-file` action, which creates an empty `_SUCCESS` file in
+each partition directory.
+
 **Syntax**
 
 ```sql

docs/docs/spark/format-table.md

@@ -112,6 +112,42 @@ change the table's managed/external ownership.
 
 ## Writes and Maintenance
 
+### Partition Mark Done
+
+With `'partition.end-input-to-done' = 'true'`, a Spark write creates an empty `_SUCCESS` file in
+each partition it writes, as Hadoop jobs do, so that downstream jobs waiting for the file can
+start. Unpartitioned tables are not marked: no `_SUCCESS` is written at the table root, and the
+procedure rejects them. A Format Table only supports the default `partition.mark-done-action`,
+`success-file`; any other action fails the statement before data is written. Only the default
+`format-table.implementation = paimon` marks partitions: writes to a table that uses `engine` do
+not, and the procedure rejects it.
+
+```sql
+CREATE TABLE my_table (id INT, dt STRING) USING parquet PARTITIONED BY (dt)
+TBLPROPERTIES ('partition.end-input-to-done' = 'true');
+
+-- creates dt=20240101/_SUCCESS, even when the query returns no rows
+INSERT OVERWRITE my_table PARTITION (dt = '20240101') SELECT id FROM source;
+```
+
+- An `INSERT OVERWRITE` that names every partition column marks that partition even without rows;
+  an `INSERT INTO` that writes no rows marks nothing.
+- Before a write or `TRUNCATE` changes a partition, its `_SUCCESS` is deleted, including a
+  partition an overwrite clears without writing to. After the commit, the write creates it again
+  in the partitions it wrote, so a partition that was only cleared is left without it. If marking
+  fails after the commit, the statement fails with the data written; mark the partitions with the
+  [mark_partition_done](./procedures/metadata#mark_partition_done) procedure.
+- Without `partition.end-input-to-done`, writes leave `_SUCCESS` files alone. When marking
+  partitions with the procedure, enable it as well, so that a later write does not leave a stale
+  file next to changed data.
+- A write marks a partition done when that write finishes, so a partition must have one writer at
+  a time. Writers are not coordinated: with two concurrent writes to a partition, the first to
+  finish marks it while the other may still be publishing files.
+- The file follows `format-table.partition-path-only-value` and is not read as data. A
+  catalog-managed partition at a custom location, or not registered, is skipped.
+- Flink writes delete the file of the partitions they change but do not create it; run the
+  procedure after them.
+
 ### Partition File Formats
 
 Catalog-managed partitions can override the table's `file.format` through the partition `options`

docs/docs/spark/procedures/metadata.md

@@ -88,6 +88,9 @@ Mark partitions as done.
 - `table` (`STRING`, required): the target table identifier.
 - `partitions` (`STRING`, required): partitions need to be mark done, If you specify multiple partitions, delimiter is ';'.
 
+A Format Table only supports the `success-file` action; see
+[Format Table Partition Mark Done](../format-table#partition-mark-done).
+
 ```sql
 -- mark single partition done
 CALL sys.mark_partition_done(table => 'default.T', partitions => 'day=2024-07-01');

paimon-core/src/main/java/org/apache/paimon/partition/actions/PartitionMarkDoneAction.java

@@ -20,8 +20,10 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FormatTable;
 import org.apache.paimon.table.PartitionMarkDone;
 import org.apache.paimon.table.PartitionModification;
+import org.apache.paimon.table.format.FormatTableSuccessFileMarkDoneAction;
 import org.apache.paimon.utils.StringUtils;
 
 import java.io.Closeable;
@@ -42,6 +44,13 @@ default void open(FileStoreTable fileStoreTable, CoreOptions options) {}
 
     void markDone(String partition) throws Exception;
 
+    /** Marks several partitions done; an implementation may batch them. */
+    default void markDone(List<String> partitions) throws Exception {
+        for (String partition : partitions) {
+            markDone(partition);
+        }
+    }
+
     static List<PartitionMarkDoneAction> createActions(
             ClassLoader cl, FileStoreTable fileStoreTable, CoreOptions options) {
         return options.partitionMarkDoneActions().stream()
@@ -82,6 +91,29 @@ static List<PartitionMarkDoneAction> createActions(
                 .collect(Collectors.toList());
     }
 
+    /** Creates the actions of a {@link FormatTable}, which only supports {@code success-file}. */
+    static List<PartitionMarkDoneAction> createActions(
+            ClassLoader cl, FormatTable formatTable, CoreOptions options) {
+        return options.partitionMarkDoneActions().stream()
+                .map(
+                        action -> {
+                            if (action != CoreOptions.PartitionMarkDoneAction.SUCCESS_FILE) {
+                                throw new UnsupportedOperationException(
+                                        String.format(
+                                                "Format Table %s only supports the '%s' partition "
+                                                        + "mark done action, but '%s' is "
+                                                        + "configured in %s.",
+                                                formatTable.fullName(),
+                                                CoreOptions.PartitionMarkDoneAction.SUCCESS_FILE,
+                                                action,
+                                                PARTITION_MARK_DONE_ACTION.key()));
+                            }
+                            return (PartitionMarkDoneAction)
+                                    new FormatTableSuccessFileMarkDoneAction(formatTable);
+                        })
+                .collect(Collectors.toList());
+    }
+
     static PartitionMarkDoneAction generateCustomMarkDoneAction(
             ClassLoader cl, CoreOptions options) {
         if (StringUtils.isNullOrWhitespaceOnly(options.partitionMarkDoneCustomClass())) {

paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java

@@ -86,22 +86,28 @@ public BatchTableCommit newCommit() {
                 table.partitionManager() != null && !table.partitionKeys().isEmpty()
                         ? options.formatTableCommitPublishThreadNum()
                         : 1;
-        return new FormatTableCommit(
-                table.location(),
-                table.partitionKeys(),
-                table.fileIO(),
-                formatTablePartitionOnlyValueInPath,
-                table.defaultPartName(),
-                overwrite,
-                Identifier.fromString(table.fullName()),
-                staticPartition,
-                syncHiveUri,
-                table.catalogContext(),
-                table.partitionManager(),
-                options.dynamicPartitionOverwrite(),
-                cleanupThreadNum,
-                publishThreadNum,
-                FormatTablePartitionOptions.fileFormat(table.options(), null));
+        FormatTableCommit commit =
+                new FormatTableCommit(
+                        table.location(),
+                        table.partitionKeys(),
+                        table.fileIO(),
+                        formatTablePartitionOnlyValueInPath,
+                        table.defaultPartName(),
+                        overwrite,
+                        Identifier.fromString(table.fullName()),
+                        staticPartition,
+                        syncHiveUri,
+                        table.catalogContext(),
+                        table.partitionManager(),
+                        options.dynamicPartitionOverwrite(),
+                        cleanupThreadNum,
+                        publishThreadNum,
+                        FormatTablePartitionOptions.fileFormat(table.options(), null));
+        // A partition marked done must not keep its success file while its data changes.
+        return commit.withSuccessFileDeletion(
+                !table.partitionKeys().isEmpty()
+                        && options.toConfiguration()
+                                .get(CoreOptions.PARTITION_MARK_DONE_WHEN_END_INPUT));
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java

@@ -33,6 +33,7 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.Partition;
 import org.apache.paimon.partition.PartitionStatistics;
+import org.apache.paimon.partition.actions.SuccessFileMarkDoneAction;
 import org.apache.paimon.stats.Statistics;
 import org.apache.paimon.table.sink.BatchTableCommit;
 import org.apache.paimon.table.sink.CommitMessage;
@@ -104,6 +105,9 @@ public class FormatTableCommit implements BatchTableCommit {
     private final int publishThreadNum;
     @Nullable private final String fileFormat;
 
+    private List<Map<String, String>> committedPartitions = Collections.emptyList();
+    private boolean deletesSuccessFiles = false;
+
     /**
      * Creates a legacy committer whose write format is unknown. Appends and overwrites reject
      * partitions with an explicit {@code file.format}, including values stored by earlier
@@ -207,6 +211,7 @@ public FormatTableCommit(
 
     @Override
     public void commit(List<CommitMessage> commitMessages) {
+        committedPartitions = Collections.emptyList();
         try {
             // One reading for the whole commit: stat-ing each file back costs a request per
             // file for a coarser number.
@@ -237,6 +242,16 @@ public void commit(List<CommitMessage> commitMessages) {
             if (overwrite && partitionManager != null) {
                 targetPartitions.forEach(partition -> reportTargetSpecs.add(partition.spec()));
             }
+            if (!overwrite && deletesSuccessFiles) {
+                Set<Path> writtenPartitionPaths = new LinkedHashSet<>();
+                for (TwoPhaseCommitMessage message : messages) {
+                    Path targetPath = message.getCommitter().targetPath();
+                    if (targetPath != null) {
+                        writtenPartitionPaths.add(targetPath.getParent());
+                    }
+                }
+                deleteSuccessFiles(new ArrayList<>(writtenPartitionPaths), cleanupThreadNum);
+            }
             Path staticPartitionPath = null;
 
             if (staticPartitions != null && !staticPartitions.isEmpty()) {
@@ -380,7 +395,7 @@ public void commit(List<CommitMessage> commitMessages) {
                     }
                 }
             }
-
+            committedPartitions = committedPartitionSpecs(writtenPartitionSpecs);
         } catch (Throwable failure) {
             // Cleanup restores the caller's interrupt before failing. Clear it only while aborting
             // staging output, then restore it; an abort failure is secondary to the commit failure
@@ -499,6 +514,40 @@ private List<Partition> loadCommitTargetPartitions(
                 : loadPartitionsByNames(writtenPartitionSpecs);
     }
 
+    /**
+     * Partitions written by the last successful commit, including the full static partition of an
+     * overwrite that wrote no rows. Empty after a failed commit or for an unpartitioned table.
+     */
+    public List<Map<String, String>> committedPartitions() {
+        return committedPartitions;
+    }
+
+    /**
+     * Whether to delete the success file of a partition before writing to it or clearing it, so
+     * that a stale one is never seen next to the changed data.
+     */
+    FormatTableCommit withSuccessFileDeletion(boolean deletesSuccessFiles) {
+        this.deletesSuccessFiles = deletesSuccessFiles;
+        return this;
+    }
+
+    private List<Map<String, String>> committedPartitionSpecs(
+            Set<Map<String, String>> writtenPartitionSpecs) {
+        if (partitionKeys == null || partitionKeys.isEmpty()) {
+            return Collections.emptyList();
+        }
+        Set<Map<String, String>> committed = new LinkedHashSet<>();
+        if (staticPartitions != null && staticPartitions.size() == partitionKeys.size()) {
+            LinkedHashMap<String, String> orderedStaticPartition = new LinkedHashMap<>();
+            for (String partitionKey : partitionKeys) {
+                orderedStaticPartition.put(partitionKey, staticPartitions.get(partitionKey));
+            }
+            committed.add(orderedStaticPartition);
+        }
+        committed.addAll(writtenPartitionSpecs);
+        return Collections.unmodifiableList(new ArrayList<>(committed));
+    }
+
     private Set<Map<String, String>> writtenPartitionSpecs(List<TwoPhaseCommitMessage> messages) {
         if (partitionKeys == null || partitionKeys.isEmpty()) {
             return Collections.emptySet();
@@ -914,6 +963,9 @@ private Set<Path> deletePreviousDataFile(Path partitionPath, int partitionLevels
 
     private Set<Path> deletePreviousDataFiles(
             List<Path> partitionPaths, int partitionLevels, int threadNum) throws IOException {
+        if (deletesSuccessFiles) {
+            deleteSuccessFiles(partitionDirectories(partitionPaths, partitionLevels), threadNum);
+        }
         Iterator<FileStatus> dataFiles = previousDataFiles(partitionPaths, partitionLevels);
         Set<Path> clearedPartitionPaths = new HashSet<>();
         try {
@@ -1186,6 +1238,69 @@ private static Throwable unwrapUncheckedIOException(Throwable failure) {
         return unwrapped;
     }
 
+    /** The partition directories the given number of levels below each path. */
+    private List<Path> partitionDirectories(List<Path> paths, int partitionLevels)
+            throws IOException {
+        if (partitionLevels == 0) {
+            return paths;
+        }
+        List<String> keys =
+                partitionKeys.subList(partitionKeys.size() - partitionLevels, partitionKeys.size());
+        List<Path> directories = new ArrayList<>();
+        for (Path path : paths) {
+            if (!fileIO.exists(path)) {
+                continue;
+            }
+            for (Pair<LinkedHashMap<String, String>, Path> partition :
+                    PartitionPathUtils.searchPartSpecAndPaths(
+                            fileIO,
+                            path,
+                            partitionLevels,
+                            keys,
+                            formatTablePartitionOnlyValueInPath,
+                            null,
+                            null,
+                            defaultPartName)) {
+                directories.add(partition.getRight());
+            }
+        }
+        return directories;
+    }
+
+    private void deleteSuccessFiles(List<Path> partitionPaths, int threadNum) throws IOException {
+        if (partitionPaths.isEmpty()) {
+            return;
+        }
+        try {
+            executeSideEffects(
+                    COMMIT_EXECUTOR,
+                    this::deleteSuccessFile,
+                    partitionPaths.iterator(),
+                    threadNum,
+                    ignored -> {});
+        } catch (UncheckedIOException e) {
+            throw (IOException) unwrapUncheckedIOException(e);
+        }
+    }
+
+    private List<Path> deleteSuccessFile(Path partitionPath) {
+        Path successPath = new Path(partitionPath, SuccessFileMarkDoneAction.SUCCESS_FILE_NAME);
+        try {
+            if (fileIO.delete(successPath, false) || !fileIO.exists(successPath)) {
+                return Collections.emptyList();
+            }
+        } catch (FileNotFoundException ignore) {
+            return Collections.emptyList();
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+        throw new UncheckedIOException(
+                new IOException(
+                        String.format(
+                                "Failed to delete success file %s of table %s.",
+                                successPath, tableIdentifier.getFullName())));
+    }
+
     /** Deletes one listed file and reports its parent when this commit removed the file. */
     private List<Path> deleteAndReportCleared(FileStatus file) {
         try {