feat(global-index): add checkpoint cleanup by task id

docs/source/user_guide/global_index.rst

@@ -38,6 +38,22 @@ Global indexes work on top of Data Evolution tables. To use global indexes, your
 - ``'row-tracking.enabled' = 'true'``
 - ``'data-evolution.enabled' = 'true'``
 
+Cleaning Up Build Checkpoints
+-----------------------------
+
+Checkpoint-enabled index builds remove their checkpoints after a successful build. If a process
+crashes, use ``GlobalIndexWriteTask::DeleteCheckpoint`` to remove all checkpoints left by that
+task once it is no longer running or being retried:
+
+.. code-block:: cpp
+
+   PAIMON_RETURN_NOT_OK(paimon::GlobalIndexWriteTask::DeleteCheckpoint(
+       table_path, "build-task-1", options, file_system));
+
+Pass the same table path and task id used in ``WriteIndex``. Task ids must be unique within the
+table. The cleanup recursively removes the whole task directory, other tasks' directories and completed
+index files are preserved.
+
 Bitmap Index
 ------------
 

include/paimon/global_index/global_index_write_task.h

@@ -27,6 +27,7 @@
 #include "paimon/global_index/indexed_split.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/result.h"
+#include "paimon/status.h"
 #include "paimon/utils/range.h"
 #include "paimon/visibility.h"
 
@@ -48,9 +49,9 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
     /// @param options      Index-specific configuration (e.g., false positive rate for bloom
     ///                     filters).
     /// @param task_id      When checkpoints are enabled, the caller must provide a non-empty task
-    ///                     identifier that uniquely identifies an index build task. Reuse it when
-    ///                     retrying the same build. If the source data, build configuration, or
-    ///                     build source code changes, the caller must use a new identifier;
+    ///                     identifier that uniquely identifies an index build task within the
+    ///                     table. Reuse it when retrying the same build. If the source data, build
+    ///                     configuration, or build source code changes, use a new identifier;
     ///                     otherwise, the index build may fail. Pass nullopt when checkpoints are
     ///                     disabled. Index types without checkpoint support ignore this value.
     /// @param pool         Memory pool for temporary allocations during index construction.
@@ -65,6 +66,19 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
         const std::map<std::string, std::string>& options,
         const std::optional<std::string>& task_id, const std::shared_ptr<MemoryPool>& pool,
         const std::shared_ptr<FileSystem>& file_system = nullptr);
+
+    /// Deletes all checkpoints left by a global index build task. The caller must ensure that this
+    /// task is no longer building or being retried.
+    ///
+    /// @param table_path   Path to the table root directory used for the build.
+    /// @param task_id      Non-empty, unique task identifier passed to `WriteIndex`.
+    /// @param options      Configuration overrides. Used to load the schema, then merged with
+    ///                     schema options, with these overrides taking precedence.
+    /// @param file_system  Specifies the file system. If `nullptr`, use default file system.
+    /// @return OK if the task directory was deleted or does not exist.
+    static Status DeleteCheckpoint(const std::string& table_path, const std::string& task_id,
+                                   const std::map<std::string, std::string>& options,
+                                   const std::shared_ptr<FileSystem>& file_system = nullptr);
 };
 
 }  // namespace paimon

src/paimon/core/global_index/global_index_file_manager_test.cpp

@@ -20,15 +20,19 @@
 #include "paimon/core/global_index/global_index_file_manager.h"
 
 #include <limits>
+#include <map>
 #include <set>
 #include <vector>
 
 #include "arrow/type.h"
 #include "fmt/format.h"
 #include "gtest/gtest.h"
 #include "paimon/common/utils/string_utils.h"
+#include "paimon/core/schema/table_schema.h"
 #include "paimon/core/utils/file_store_path_factory.h"
+#include "paimon/defs.h"
 #include "paimon/fs/local/local_file_system.h"
+#include "paimon/global_index/global_index_write_task.h"
 #include "paimon/testing/utils/testharness.h"
 #include "paimon/utils/range.h"
 
@@ -46,8 +50,24 @@ class GlobalIndexFileManagerTest : public ::testing::Test {
             }
             return LocalFileSystem::ListDir(directory, status_list);
         }
+        Status Delete(const std::string& path, bool recursive = true) const override {
+            delete_paths_.push_back(path);
+            if (path == fail_delete_path_) {
+                return Status::IOError("checkpoint delete failed");
+            }
+            return LocalFileSystem::Delete(path, recursive);
+        }
+        Result<bool> Exists(const std::string& path) const override {
+            if (path == fail_exists_path_) {
+                return Status::IOError("checkpoint exists failed");
+            }
+            return LocalFileSystem::Exists(path);
+        }
         bool fail_list_ = false;
+        std::string fail_delete_path_;
+        std::string fail_exists_path_;
         mutable int32_t list_count_ = 0;
+        mutable std::vector<std::string> delete_paths_;
     };
 
     void SetUp() override {
@@ -80,6 +100,17 @@ class GlobalIndexFileManagerTest : public ::testing::Test {
         return PathUtil::GetName(uri);
     }
 
+    Status CreateSchema(const std::map<std::string, std::string>& options = {}) const {
+        PAIMON_ASSIGN_OR_RAISE(
+            std::unique_ptr<TableSchema> schema,
+            TableSchema::Create(/*schema_id=*/0,
+                                arrow::schema({arrow::field("vector", arrow::int32())}),
+                                /*partition_keys=*/{}, /*primary_keys=*/{}, options));
+        PAIMON_ASSIGN_OR_RAISE(std::string content, schema->ToJsonString());
+        return fs_->WriteFile(PathUtil::JoinPath(dir_->Str(), "schema/schema-0"), content,
+                              /*overwrite=*/false);
+    }
+
     std::shared_ptr<TrackingFileSystem> fs_ = std::make_shared<TrackingFileSystem>();
     std::unique_ptr<UniqueTestDirectory> dir_;
     std::shared_ptr<FileStorePathFactory> path_factory_;
@@ -140,8 +171,8 @@ TEST_F(GlobalIndexFileManagerTest, TestCheckpointNotConfigured) {
 TEST_F(GlobalIndexFileManagerTest, TestCheckpointIOAndRestart) {
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager, CreateManager());
     ASSERT_EQ(fs_->list_count_, 0);
-    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint");
-    std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1");
+    std::string prefix = "lumina-global-index-vector-10-20-";
     std::string previous_path = PathUtil::JoinPath(directory, prefix + "9.index.ckpt");
     // The latest sequence must be discovered on first file name allocation, not at construction.
     ASSERT_OK(fs_->WriteFile(previous_path, "old", /*overwrite=*/false));
@@ -167,8 +198,8 @@ TEST_F(GlobalIndexFileManagerTest, TestCheckpointIOAndRestart) {
     ASSERT_OK_AND_ASSIGN(name, CreateCheckpointFile(restarted));
     ASSERT_TRUE(StringUtils::EndsWith(name, "-12.index.ckpt"));
 
-    std::string other_task_path =
-        PathUtil::JoinPath(directory, "lumina-global-index-vector-10-20-task-2-99.index.ckpt");
+    std::string other_task_path = PathUtil::JoinPath(
+        dir_->Str(), "index/checkpoint/task-2/lumina-global-index-vector-10-20-99.index.ckpt");
     std::string index_path = manager->ToPath("retained.index");
     ASSERT_OK(fs_->WriteFile(other_task_path, "other", /*overwrite=*/false));
     ASSERT_OK(fs_->WriteFile(index_path, "index", /*overwrite=*/false));
@@ -205,8 +236,8 @@ TEST_F(GlobalIndexFileManagerTest, TestInitializationFailureCanRetry) {
 
 TEST_F(GlobalIndexFileManagerTest, TestLatestCheckpointUsesNumericId) {
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager, CreateManager());
-    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint");
-    std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1");
+    std::string prefix = "lumina-global-index-vector-10-20-";
     ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix + "9.index.ckpt"), "older",
                              /*overwrite=*/false));
     ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix + "10.index.ckpt"), "newer",
@@ -223,8 +254,8 @@ TEST_F(GlobalIndexFileManagerTest, TestReadsDoNotInitializeFileId) {
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager, CreateManager());
     ASSERT_OK_AND_ASSIGN(bool exists, manager->CheckpointExists());
     ASSERT_FALSE(exists);
-    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint");
-    std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1");
+    std::string prefix = "lumina-global-index-vector-10-20-";
     ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix + "9.index.ckpt"), "old",
                              /*overwrite=*/false));
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input, manager->OpenCheckpointInputStream());
@@ -243,8 +274,8 @@ TEST_F(GlobalIndexFileManagerTest, TestReadsDoNotInitializeFileId) {
 
 TEST_F(GlobalIndexFileManagerTest, TestFileIdOverflow) {
     ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager, CreateManager());
-    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint");
-    std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1");
+    std::string prefix = "lumina-global-index-vector-10-20-";
     int64_t max_id = std::numeric_limits<int64_t>::max();
     ASSERT_OK(fs_->WriteFile(
         PathUtil::JoinPath(directory, fmt::format("{}{}.index.ckpt", prefix, max_id - 1)), "old",
@@ -291,4 +322,102 @@ TEST_F(GlobalIndexFileManagerTest, TestDeleteDoesNotResetFileId) {
     ASSERT_EQ(fs_->list_count_, 2);
 }
 
+TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpoint) {
+    ASSERT_OK(CreateSchema());
+    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1");
+    std::vector<std::string> matching_names = {"lumina-global-index-vector-10-20-9.index.ckpt",
+                                               "lumina-global-index-vector-10-20-10.index.ckpt",
+                                               "lumina-global-index-other-10-20-0.index.ckpt",
+                                               "other-global-index-vector-10-20-0.index.ckpt",
+                                               "lumina-global-index-vector-10-21-0.index.ckpt",
+                                               "lumina-global-index-vector-11-20-0.index.ckpt",
+                                               "lumina-global-index-vector-10-20-0.index.ckpt.bak",
+                                               "incomplete-file",
+                                               "nested/unfinished-file"};
+    for (const std::string& name : matching_names) {
+        ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, name), "incomplete checkpoint",
+                                 /*overwrite=*/false));
+    }
+    std::vector<std::string> retained_tasks = {"task-2", "task-1-extended", "task-1-9", "task-1-"};
+    std::vector<std::string> retained_paths;
+    for (const std::string& task_id : retained_tasks) {
+        std::string path = PathUtil::JoinPath(
+            dir_->Str(), "index/checkpoint/" + task_id + "/" + matching_names.front());
+        ASSERT_OK(fs_->WriteFile(path, "retained", /*overwrite=*/false));
+        retained_paths.push_back(path);
+    }
+    ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_));
+    ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(directory));
+    ASSERT_FALSE(exists);
+    ASSERT_EQ(fs_->delete_paths_, std::vector<std::string>({directory}));
+    for (const std::string& path : retained_paths) {
+        ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(path));
+        ASSERT_TRUE(exists) << path;
+    }
+    ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_));
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointWithoutCustomFileSystem) {
+    ASSERT_OK(CreateSchema());
+    ASSERT_NOK_WITH_MSG(GlobalIndexWriteTask::DeleteCheckpoint(
+                            dir_->Str(), "task-1", {{Options::FILE_SYSTEM, "unregistered-fs"}}),
+                        "Create factory failed with identifier 'unregistered-fs'");
+    ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}));
+    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1");
+    ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(directory));
+    ASSERT_FALSE(exists);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager, CreateManager());
+    ASSERT_OK(CreateCheckpointFile(manager));
+    ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}));
+    ASSERT_OK_AND_ASSIGN(exists, manager->CheckpointExists());
+    ASSERT_FALSE(exists);
+    ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}));
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointRejectsInvalidTaskId) {
+    for (const std::string& invalid : std::vector<std::string>{"", ".", "..", " ", "a/b", "a\\b",
+                                                               "a\nb", std::string("a\0b", 3)}) {
+        ASSERT_NOK(
+            GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), invalid, /*options=*/{}, fs_));
+    }
+    ASSERT_EQ(fs_->list_count_, 0);
+    ASSERT_TRUE(fs_->delete_paths_.empty());
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointPropagatesFileSystemErrors) {
+    ASSERT_OK(CreateSchema());
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager, CreateManager());
+    ASSERT_OK_AND_ASSIGN(std::string first, CreateCheckpointFile(manager));
+    ASSERT_OK_AND_ASSIGN(std::string second, CreateCheckpointFile(manager));
+    fs_->fail_list_ = true;
+    ASSERT_NOK_WITH_MSG(
+        GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_),
+        "checkpoint list failed");
+    ASSERT_TRUE(fs_->delete_paths_.empty());
+
+    fs_->fail_list_ = false;
+    std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1");
+    fs_->fail_exists_path_ = directory;
+    ASSERT_NOK_WITH_MSG(
+        GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_),
+        "checkpoint exists failed");
+    ASSERT_TRUE(fs_->delete_paths_.empty());
+    fs_->fail_exists_path_.clear();
+    fs_->fail_delete_path_ = directory;
+    ASSERT_NOK_WITH_MSG(
+        GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_),
+        "checkpoint delete failed");
+    ASSERT_EQ(fs_->delete_paths_, std::vector<std::string>({directory}));
+    for (const std::string& name : {first, second}) {
+        ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(PathUtil::JoinPath(directory, name)));
+        ASSERT_TRUE(exists);
+    }
+
+    fs_->fail_delete_path_.clear();
+    ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_));
+    ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(directory));
+    ASSERT_FALSE(exists);
+}
+
 }  // namespace paimon::test