feat(global-index): support checkpoints for Lumina index builds

include/paimon/global_index/global_index_write_task.h

@@ -21,6 +21,7 @@
 
 #include <map>
 #include <memory>
+#include <optional>
 #include <string>
 
 #include "paimon/global_index/indexed_split.h"
@@ -45,7 +46,13 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
     ///                     The range must be fully contained within the data covered
     ///                     by the given `indexed_split`.
     /// @param options      Index-specific configuration (e.g., false positive rate for bloom
-    /// filters).
+    ///                     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;
+    ///                     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.
     ///                     If `nullptr`, the system's default memory pool will be used.
     /// @param file_system  Specifies the file system for file operations.
@@ -55,7 +62,8 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
     static Result<std::shared_ptr<CommitMessage>> WriteIndex(
         const std::string& table_path, const std::string& field_name, const std::string& index_type,
         const std::shared_ptr<IndexedSplit>& indexed_split,
-        const std::map<std::string, std::string>& options, const std::shared_ptr<MemoryPool>& pool,
+        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);
 };
 

include/paimon/global_index/global_indexer.h

@@ -70,6 +70,11 @@ class PAIMON_EXPORT GlobalIndexer {
         ::ArrowSchema* arrow_schema, const std::shared_ptr<GlobalIndexFileReader>& file_reader,
         const std::vector<GlobalIndexIOMeta>& files,
         const std::shared_ptr<MemoryPool>& pool) const = 0;
+
+    /// Whether this indexer supports checkpointing an index build.
+    virtual bool SupportsCheckpoint() const {
+        return false;
+    }
 };
 
 }  // namespace paimon

include/paimon/global_index/io/global_index_checkpoint_file_manager.h

@@ -0,0 +1,53 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+class InputStream;
+class OutputStream;
+
+/// Abstract interface for managing checkpoints belonging to one global index build identity.
+class PAIMON_EXPORT GlobalIndexCheckpointFileManager {
+ public:
+    virtual ~GlobalIndexCheckpointFileManager() = default;
+
+    /// Returns whether checkpoint storage is configured, without accessing storage.
+    virtual bool SupportsCheckpoint() const = 0;
+
+    /// Creates a new checkpoint file and opens it for writing.
+    virtual Result<std::unique_ptr<OutputStream>> CreateCheckpointOutputStream() const = 0;
+
+    /// Opens the matching checkpoint with the largest numeric id for reading.
+    virtual Result<std::unique_ptr<InputStream>> OpenCheckpointInputStream() const = 0;
+
+    /// Returns whether the checkpoint file exists.
+    virtual Result<bool> CheckpointExists() const = 0;
+
+    /// Deletes all matching checkpoint files. Deleting missing checkpoints succeeds.
+    virtual Status DeleteCheckpoint() const = 0;
+};
+
+}  // namespace paimon

src/paimon/CMakeLists.txt

@@ -813,6 +813,7 @@ if(PAIMON_BUILD_TESTS)
                     core/io/file_index_evaluator_test.cpp
                     core/io/single_file_writer_test.cpp
                     core/io/rolling_blob_file_writer_test.cpp
+                    core/global_index/global_index_file_manager_test.cpp
                     core/global_index/global_index_evaluator_impl_test.cpp
                     core/global_index/indexed_split_test.cpp
                     core/manifest/file_source_test.cpp

src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp

@@ -67,7 +67,8 @@ class BitmapGlobalIndexTest : public ::testing::Test {
         auto global_index = std::make_shared<BitmapGlobalIndex>(file_index);
 
         auto path_factory = std::make_shared<MockIndexPathFactory>(index_root);
-        auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_, path_factory);
+        auto file_writer = std::make_shared<GlobalIndexFileManager>(
+            fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
 
         PAIMON_ASSIGN_OR_RAISE(
             std::shared_ptr<GlobalIndexWriter> global_writer,
@@ -112,7 +113,8 @@ class BitmapGlobalIndexTest : public ::testing::Test {
         auto global_index = std::make_shared<BitmapGlobalIndex>(file_index);
 
         auto path_factory = std::make_shared<MockIndexPathFactory>(index_root);
-        auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_, path_factory);
+        auto file_reader = std::make_shared<GlobalIndexFileManager>(
+            fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
         EXPECT_OK_AND_ASSIGN(
             auto global_index_reader,
             global_index->CreateReader(CreateArrowSchema(type).get(), file_reader, {meta}, pool_));

src/paimon/common/global_index/global_indexer_factory_test.cpp

@@ -39,6 +39,7 @@ TEST(GlobalIndexerFactoryTest, TestLegacyBitmapEnabledForTesting) {
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<GlobalIndexer> indexer,
                          GlobalIndexerFactory::Get("bitmap", options));
     ASSERT_TRUE(dynamic_cast<BitmapGlobalIndex*>(indexer.get()));
+    ASSERT_FALSE(indexer->SupportsCheckpoint());
 }
 
 TEST(GlobalIndexerFactoryTest, TestNonExist) {

src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp

@@ -66,7 +66,8 @@ class RangeBitmapGlobalIndexTest : public ::testing::Test {
         auto global_index = std::make_shared<RangeBitmapGlobalIndex>(file_index);
 
         auto path_factory = std::make_shared<MockIndexPathFactory>(index_root);
-        auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_, path_factory);
+        auto file_writer = std::make_shared<GlobalIndexFileManager>(
+            fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
 
         PAIMON_ASSIGN_OR_RAISE(
             std::shared_ptr<GlobalIndexWriter> global_writer,
@@ -109,7 +110,8 @@ class RangeBitmapGlobalIndexTest : public ::testing::Test {
         auto global_index = std::make_shared<RangeBitmapGlobalIndex>(file_index);
 
         auto path_factory = std::make_shared<MockIndexPathFactory>(index_root);
-        auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_, path_factory);
+        auto file_reader = std::make_shared<GlobalIndexFileManager>(
+            fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
         EXPECT_OK_AND_ASSIGN(
             auto global_index_reader,
             global_index->CreateReader(CreateArrowSchema(type).get(), file_reader, {meta}, pool_));

src/paimon/core/global_index/global_index_file_manager.h

@@ -19,22 +19,37 @@
 
 #pragma once
 
+#include <algorithm>
+#include <cstdint>
+#include <limits>
 #include <memory>
+#include <optional>
 #include <string>
+#include <utility>
+#include <vector>
 
+#include "paimon/common/utils/path_util.h"
 #include "paimon/common/utils/uuid.h"
+#include "paimon/core/index/index_checkpoint_path_factory.h"
 #include "paimon/core/index/index_path_factory.h"
 #include "paimon/fs/file_system.h"
+#include "paimon/global_index/io/global_index_checkpoint_file_manager.h"
 #include "paimon/global_index/io/global_index_file_reader.h"
 #include "paimon/global_index/io/global_index_file_writer.h"
 
 namespace paimon {
 /// Helper class for managing global index files.
-class GlobalIndexFileManager : public GlobalIndexFileReader, public GlobalIndexFileWriter {
+/// Checkpoint storage is optional and is never accessed by construction or ordinary index I/O.
+class GlobalIndexFileManager : public GlobalIndexFileReader,
+                               public GlobalIndexFileWriter,
+                               public GlobalIndexCheckpointFileManager {
  public:
     GlobalIndexFileManager(const std::shared_ptr<FileSystem>& fs,
-                           const std::shared_ptr<IndexPathFactory>& path_factory)
-        : fs_(fs), path_factory_(path_factory) {}
+                           const std::shared_ptr<IndexPathFactory>& path_factory,
+                           std::unique_ptr<IndexCheckpointPathFactory> checkpoint_path_factory)
+        : fs_(fs),
+          path_factory_(path_factory),
+          checkpoint_path_factory_(std::move(checkpoint_path_factory)) {}
 
     Result<std::unique_ptr<InputStream>> GetInputStream(
         const std::string& file_path) const override {
@@ -71,8 +86,110 @@ class GlobalIndexFileManager : public GlobalIndexFileReader, public GlobalIndexF
         return path_factory_->IsExternalPath();
     }
 
+    bool SupportsCheckpoint() const override {
+        return checkpoint_path_factory_ != nullptr;
+    }
+
+    Result<std::unique_ptr<OutputStream>> CreateCheckpointOutputStream() const override {
+        if (!SupportsCheckpoint()) {
+            return Status::Invalid("global index checkpoint storage is not configured");
+        }
+        PAIMON_ASSIGN_OR_RAISE(int64_t file_id, NextCheckpointFileId());
+        PAIMON_RETURN_NOT_OK(fs_->Mkdirs(checkpoint_path_factory_->GetDirectoryPath()));
+        return fs_->Create(checkpoint_path_factory_->NewPath(file_id), /*overwrite=*/false);
+    }
+
+    Result<std::unique_ptr<InputStream>> OpenCheckpointInputStream() const override {
+        if (!SupportsCheckpoint()) {
+            return Status::Invalid("global index checkpoint storage is not configured");
+        }
+        PAIMON_ASSIGN_OR_RAISE(std::optional<CheckpointFile> checkpoint_file,
+                               LatestCheckpointFile());
+        if (!checkpoint_file) {
+            return Status::NotExist("global index checkpoint file does not exist");
+        }
+        return fs_->Open(checkpoint_file->path);
+    }
+
+    Result<bool> CheckpointExists() const override {
+        if (!SupportsCheckpoint()) {
+            return Status::Invalid("global index checkpoint storage is not configured");
+        }
+        PAIMON_ASSIGN_OR_RAISE(std::optional<CheckpointFile> checkpoint_file,
+                               LatestCheckpointFile());
+        return checkpoint_file.has_value();
+    }
+
+    Status DeleteCheckpoint() const override {
+        if (!SupportsCheckpoint()) {
+            return Status::Invalid("global index checkpoint storage is not configured");
+        }
+        PAIMON_ASSIGN_OR_RAISE(std::vector<CheckpointFile> checkpoint_files, ListCheckpointFiles());
+        Status first_error = Status::OK();
+        for (const CheckpointFile& checkpoint_file : checkpoint_files) {
+            Status status = fs_->Delete(checkpoint_file.path, /*recursive=*/false);
+            if (!status.ok() && first_error.ok()) {
+                first_error = std::move(status);
+            }
+        }
+        return first_error;
+    }
+
  private:
+    struct CheckpointFile {
+        int64_t id;
+        std::string path;
+    };
+
+    Result<int64_t> NextCheckpointFileId() const {
+        if (!last_checkpoint_file_id_) {
+            PAIMON_ASSIGN_OR_RAISE(std::optional<CheckpointFile> checkpoint_file,
+                                   LatestCheckpointFile());
+            last_checkpoint_file_id_ = checkpoint_file ? checkpoint_file->id : -1;
+        }
+        if (last_checkpoint_file_id_.value() == std::numeric_limits<int64_t>::max()) {
+            return Status::Invalid("checkpoint file id exceeds int64 max");
+        }
+        return ++last_checkpoint_file_id_.value();
+    }
+
+    Result<std::vector<CheckpointFile>> ListCheckpointFiles() const {
+        std::vector<BasicFileStatus> file_statuses;
+        PAIMON_RETURN_NOT_OK(
+            fs_->ListDir(checkpoint_path_factory_->GetDirectoryPath(), &file_statuses));
+        std::vector<CheckpointFile> checkpoint_files;
+        for (const BasicFileStatus& file_status : file_statuses) {
+            if (file_status.IsDir()) {
+                continue;
+            }
+            std::string file_name = PathUtil::GetName(file_status.GetPath());
+            std::optional<int64_t> id = checkpoint_path_factory_->GetCheckpointId(file_name);
+            if (!id) {
+                continue;
+            }
+            checkpoint_files.push_back(
+                CheckpointFile{id.value(), checkpoint_path_factory_->ToPath(file_name)});
+        }
+        return checkpoint_files;
+    }
+
+    Result<std::optional<CheckpointFile>> LatestCheckpointFile() const {
+        PAIMON_ASSIGN_OR_RAISE(std::vector<CheckpointFile> checkpoint_files, ListCheckpointFiles());
+        if (checkpoint_files.empty()) {
+            return std::optional<CheckpointFile>();
+        }
+        CheckpointFile latest =
+            *std::max_element(checkpoint_files.begin(), checkpoint_files.end(),
+                              [](const CheckpointFile& left, const CheckpointFile& right) {
+                                  return left.id < right.id;
+                              });
+        return std::optional<CheckpointFile>(std::move(latest));
+    }
+
     std::shared_ptr<FileSystem> fs_;
     std::shared_ptr<IndexPathFactory> path_factory_;
+    std::unique_ptr<IndexCheckpointPathFactory> checkpoint_path_factory_;
+    // Historical ids are loaded only on the first successful file id allocation scan.
+    mutable std::optional<int64_t> last_checkpoint_file_id_;
 };
 }  // namespace paimon