This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new a774f8e6 feat(global-index): add checkpoint cleanup by task id (#465)
a774f8e6 is described below
commit a774f8e6be40579ccfe10ac6c72b72896c704c61
Author: lszskye <[email protected]>
AuthorDate: Tue Oct 27 08:51:53 2026 +0800
feat(global-index): add checkpoint cleanup by task id (#465)
---
docs/source/user_guide/global_index.rst | 16 +++
.../paimon/global_index/global_index_write_task.h | 20 ++-
.../global_index_file_manager_test.cpp | 149 +++++++++++++++++++--
.../core/global_index/global_index_write_task.cpp | 60 ++++++---
.../core/index/index_checkpoint_path_factory.h | 11 ++
src/paimon/core/utils/file_store_path_factory.cpp | 8 +-
.../core/utils/file_store_path_factory_test.cpp | 24 +++-
.../indexer/lumina/lumina_global_index_test.cpp | 9 +-
test/inte/global_index_test.cpp | 23 +++-
9 files changed, 269 insertions(+), 51 deletions(-)
diff --git a/docs/source/user_guide/global_index.rst
b/docs/source/user_guide/global_index.rst
index f3dca744..897cfebb 100644
--- a/docs/source/user_guide/global_index.rst
+++ b/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
------------
diff --git a/include/paimon/global_index/global_index_write_task.h
b/include/paimon/global_index/global_index_write_task.h
index 391c02ca..8c89a9c1 100644
--- a/include/paimon/global_index/global_index_write_task.h
+++ b/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
diff --git a/src/paimon/core/global_index/global_index_file_manager_test.cpp
b/src/paimon/core/global_index/global_index_file_manager_test.cpp
index ab27ad2e..d5cce4ad 100644
--- a/src/paimon/core/global_index/global_index_file_manager_test.cpp
+++ b/src/paimon/core/global_index/global_index_file_manager_test.cpp
@@ -20,6 +20,7 @@
#include "paimon/core/global_index/global_index_file_manager.h"
#include <limits>
+#include <map>
#include <set>
#include <vector>
@@ -27,8 +28,11 @@
#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
diff --git a/src/paimon/core/global_index/global_index_write_task.cpp
b/src/paimon/core/global_index/global_index_write_task.cpp
index d5bd13e9..f04d96bb 100644
--- a/src/paimon/core/global_index/global_index_write_task.cpp
+++ b/src/paimon/core/global_index/global_index_write_task.cpp
@@ -20,6 +20,7 @@
#include "paimon/global_index/global_index_write_task.h"
#include <set>
+#include <utility>
#include "arrow/array/array_nested.h"
#include "arrow/c/bridge.h"
@@ -34,6 +35,7 @@
#include "paimon/core/casting/casting_utils.h"
#include "paimon/core/core_options.h"
#include "paimon/core/global_index/global_index_file_manager.h"
+#include "paimon/core/index/index_checkpoint_path_factory.h"
#include "paimon/core/io/data_increment.h"
#include "paimon/core/schema/schema_manager.h"
#include "paimon/core/schema/table_schema.h"
@@ -47,6 +49,26 @@
#include "paimon/table/source/table_read.h"
namespace paimon {
namespace {
+Result<std::pair<std::shared_ptr<TableSchema>, CoreOptions>>
LoadSchemaAndOptions(
+ const std::string& table_path, const std::map<std::string, std::string>&
options,
+ const std::shared_ptr<FileSystem>& file_system) {
+ PAIMON_ASSIGN_OR_RAISE(CoreOptions tmp_options,
CoreOptions::FromMap(options, file_system));
+ SchemaManager schema_manager(tmp_options.GetFileSystem(), table_path);
+ PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>>
latest_table_schema,
+ schema_manager.Latest());
+ if (latest_table_schema == std::nullopt) {
+ return Status::Invalid("not found latest schema");
+ }
+ const auto& table_schema = latest_table_schema.value();
+ auto final_options = table_schema->Options();
+ for (const auto& [key, value] : options) {
+ final_options[key] = value;
+ }
+ PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options,
+ CoreOptions::FromMap(final_options, file_system));
+ return std::make_pair(table_schema, std::move(core_options));
+}
+
Result<std::unique_ptr<GlobalIndexer>> CreateGlobalIndexer(const std::string&
index_type,
const CoreOptions&
core_options) {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<GlobalIndexer> indexer,
@@ -342,22 +364,10 @@ Result<std::shared_ptr<CommitMessage>>
GlobalIndexWriteTask::WriteIndex(
std::shared_ptr<MemoryPool> pool = memory_pool ? memory_pool :
GetDefaultPool();
std::shared_ptr<arrow::MemoryPool> arrow_pool = GetArrowPool(pool);
- // load schema
- PAIMON_ASSIGN_OR_RAISE(CoreOptions tmp_options,
CoreOptions::FromMap(options, file_system));
- SchemaManager schema_manager(tmp_options.GetFileSystem(), table_path);
- PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>>
latest_table_schema,
- schema_manager.Latest());
- if (latest_table_schema == std::nullopt) {
- return Status::Invalid("not found latest schema");
- }
- // merge options
- const auto& table_schema = latest_table_schema.value();
- auto final_options = table_schema->Options();
- for (const auto& [key, value] : options) {
- final_options[key] = value;
- }
- PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options,
- CoreOptions::FromMap(final_options, file_system));
+ std::pair<std::shared_ptr<TableSchema>, CoreOptions> schema_and_options;
+ PAIMON_ASSIGN_OR_RAISE(schema_and_options,
+ LoadSchemaAndOptions(table_path, options,
file_system));
+ const auto& [table_schema, core_options] = schema_and_options;
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<GlobalIndexer> indexer,
CreateGlobalIndexer(index_type, core_options));
PAIMON_ASSIGN_OR_RAISE(std::optional<std::vector<std::string>>
extra_field_names,
@@ -410,4 +420,22 @@ Result<std::shared_ptr<CommitMessage>>
GlobalIndexWriteTask::WriteIndex(
extra_field_ids);
}
+Status GlobalIndexWriteTask::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) {
+ PAIMON_ASSIGN_OR_RAISE(std::string task_directory,
+
IndexCheckpointPathFactory::GetTaskDirectory(table_path, task_id));
+ std::pair<std::shared_ptr<TableSchema>, CoreOptions> schema_and_options;
+ PAIMON_ASSIGN_OR_RAISE(schema_and_options,
+ LoadSchemaAndOptions(table_path, options,
file_system));
+ std::shared_ptr<FileSystem> fs = schema_and_options.second.GetFileSystem();
+ PAIMON_ASSIGN_OR_RAISE(bool exists, fs->Exists(task_directory));
+ if (!exists) {
+ return Status::OK();
+ }
+ Status status = fs->Delete(task_directory, /*recursive=*/true);
+ return status.IsNotExist() ? Status::OK() : status;
+}
+
} // namespace paimon
diff --git a/src/paimon/core/index/index_checkpoint_path_factory.h
b/src/paimon/core/index/index_checkpoint_path_factory.h
index 0c1b13e5..0bfa56aa 100644
--- a/src/paimon/core/index/index_checkpoint_path_factory.h
+++ b/src/paimon/core/index/index_checkpoint_path_factory.h
@@ -22,6 +22,10 @@
#include <optional>
#include <string>
+#include "paimon/common/utils/path_util.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
namespace paimon {
/// Path factory for global index checkpoints scoped to one task.
@@ -38,6 +42,13 @@ class IndexCheckpointPathFactory {
/// Returns the checkpoint id when the file name belongs to this factory.
virtual std::optional<int64_t> GetCheckpointId(const std::string&
file_name) const = 0;
+
+ /// Returns the task's checkpoint directory after validating its
identifier.
+ static Result<std::string> GetTaskDirectory(const std::string& table_path,
+ const std::string& task_id) {
+ PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint
task id", task_id));
+ return PathUtil::JoinPath(PathUtil::JoinPath(table_path,
"index/checkpoint"), task_id);
+ }
};
} // namespace paimon
diff --git a/src/paimon/core/utils/file_store_path_factory.cpp
b/src/paimon/core/utils/file_store_path_factory.cpp
index ac15ad63..9acb6d4e 100644
--- a/src/paimon/core/utils/file_store_path_factory.cpp
+++ b/src/paimon/core/utils/file_store_path_factory.cpp
@@ -248,10 +248,10 @@
FileStorePathFactory::CreateGlobalIndexCheckpointPathFactory(const std::string&
};
PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint index
type", index_type));
PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint
field", field_name));
- PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint task
id", task_id));
- std::string directory = PathUtil::JoinPath(IndexPath(root_), "checkpoint");
- std::string file_name_prefix = fmt::format("{}-global-index-{}-{}-{}-{}-",
index_type,
- field_name, range.from,
range.to, task_id);
+ PAIMON_ASSIGN_OR_RAISE(std::string directory,
+ IndexCheckpointPathFactory::GetTaskDirectory(root_,
task_id));
+ std::string file_name_prefix =
+ fmt::format("{}-global-index-{}-{}-{}-", index_type, field_name,
range.from, range.to);
return std::unique_ptr<IndexCheckpointPathFactory>(
std::make_unique<IndexCheckpointPathFactoryImpl>(directory,
file_name_prefix));
}
diff --git a/src/paimon/core/utils/file_store_path_factory_test.cpp
b/src/paimon/core/utils/file_store_path_factory_test.cpp
index bfb2b962..447cb4f4 100644
--- a/src/paimon/core/utils/file_store_path_factory_test.cpp
+++ b/src/paimon/core/utils/file_store_path_factory_test.cpp
@@ -596,11 +596,14 @@ TEST_F(FileStorePathFactoryTest,
TestCreateGlobalIndexCheckpointPathFactory) {
std::shared_ptr<FileStorePathFactory> file_store_path_factory =
CreateFactory(dir->Str());
std::shared_ptr<FileSystem> file_system = dir->GetFileSystem();
- std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+ std::string prefix = "lumina-global-index-vector-10-20-";
ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_path_factory,
file_store_path_factory->CreateGlobalIndexCheckpointPathFactory(
"lumina", "vector", Range(10, 20), "task-1"));
- std::string checkpoint_dir = PathUtil::JoinPath(dir->Str(),
"index/checkpoint");
+ std::string checkpoint_dir = PathUtil::JoinPath(dir->Str(),
"index/checkpoint/task-1");
+ ASSERT_OK_AND_ASSIGN(std::string task_directory,
+
IndexCheckpointPathFactory::GetTaskDirectory(dir->Str(), "task-1"));
+ ASSERT_EQ(task_directory, checkpoint_dir);
ASSERT_EQ(checkpoint_path_factory->GetDirectoryPath(), checkpoint_dir);
std::string first_path = checkpoint_path_factory->NewPath(0);
ASSERT_EQ(first_path, PathUtil::JoinPath(checkpoint_dir, prefix +
"0.index.ckpt"));
@@ -609,9 +612,9 @@ TEST_F(FileStorePathFactoryTest,
TestCreateGlobalIndexCheckpointPathFactory) {
ASSERT_EQ(checkpoint_path_factory->ToPath("checkpoint"),
PathUtil::JoinPath(checkpoint_dir, "checkpoint"));
ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(prefix +
"9.index.ckpt"), 9);
- ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(
- "lumina-global-index-other-10-20-task-1-9.index.ckpt"),
- std::nullopt);
+ ASSERT_EQ(
+
checkpoint_path_factory->GetCheckpointId("lumina-global-index-other-10-20-9.index.ckpt"),
+ std::nullopt);
ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(prefix +
"invalid-10.index.ckpt"),
std::nullopt);
ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(prefix +
"-1.index.ckpt"), std::nullopt);
@@ -659,14 +662,18 @@ TEST_F(FileStorePathFactoryTest,
TestCheckpointTaskIsolation) {
ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexCheckpointPathFactory>
other_factory,
factory->CreateGlobalIndexCheckpointPathFactory("lumina", "vector",
Range(10, 20), "task-2"));
- ASSERT_EQ(own_factory->GetDirectoryPath(),
other_factory->GetDirectoryPath());
+ ASSERT_NE(own_factory->GetDirectoryPath(),
other_factory->GetDirectoryPath());
ASSERT_OK(fs->Mkdirs(own_factory->GetDirectoryPath()));
+ ASSERT_OK(fs->Mkdirs(other_factory->GetDirectoryPath()));
std::string own_path = own_factory->NewPath(0);
ASSERT_OK(fs->WriteFile(own_path, "checkpoint", /*overwrite=*/false));
+ std::string same_name_other_path = other_factory->NewPath(0);
+ ASSERT_EQ(PathUtil::GetName(own_path),
PathUtil::GetName(same_name_other_path));
+ ASSERT_OK(fs->WriteFile(same_name_other_path, "foreign",
/*overwrite=*/false));
std::string other_path = other_factory->NewPath(100);
ASSERT_OK(fs->WriteFile(other_path, "foreign", /*overwrite=*/false));
ASSERT_EQ(own_factory->GetCheckpointId(PathUtil::GetName(own_path)), 0);
- ASSERT_EQ(own_factory->GetCheckpointId(PathUtil::GetName(other_path)),
std::nullopt);
+ ASSERT_EQ(own_factory->GetCheckpointId(PathUtil::GetName(other_path)),
100);
GlobalIndexFileManager manager(fs, factory->CreateGlobalIndexFileFactory(),
std::move(own_factory));
@@ -690,6 +697,8 @@ TEST_F(FileStorePathFactoryTest,
TestCheckpointTaskIsolation) {
ASSERT_OK(manager.DeleteCheckpoint());
ASSERT_OK_AND_ASSIGN(exists, fs->Exists(other_path));
ASSERT_TRUE(exists);
+ ASSERT_OK_AND_ASSIGN(exists, fs->Exists(same_name_other_path));
+ ASSERT_TRUE(exists);
}
TEST_F(FileStorePathFactoryTest, TestCheckpointRejectsInvalidPathComponents) {
@@ -698,6 +707,7 @@ TEST_F(FileStorePathFactoryTest,
TestCheckpointRejectsInvalidPathComponents) {
auto factory = CreateFactory(dir->Str());
for (const std::string& invalid : std::vector<std::string>{"", ".", "..",
" ", "a/b", "a\\b",
"a\nb",
std::string("a\0b", 3)}) {
+ ASSERT_NOK(IndexCheckpointPathFactory::GetTaskDirectory(dir->Str(),
invalid));
ASSERT_NOK(factory->CreateGlobalIndexCheckpointPathFactory(invalid,
"vector", Range(10, 20),
"task"));
ASSERT_NOK(factory->CreateGlobalIndexCheckpointPathFactory("lumina",
invalid, Range(10, 20),
diff --git a/src/paimon/indexer/lumina/lumina_global_index_test.cpp
b/src/paimon/indexer/lumina/lumina_global_index_test.cpp
index e3451c44..0d5605f4 100644
--- a/src/paimon/indexer/lumina/lumina_global_index_test.cpp
+++ b/src/paimon/indexer/lumina/lumina_global_index_test.cpp
@@ -429,7 +429,7 @@ TEST_F(LuminaGlobalIndexTest, TestCheckpointFileManagement)
{
auto test_root_dir = paimon::test::UniqueTestDirectory::Create();
ASSERT_TRUE(test_root_dir);
std::string test_root = test_root_dir->Str();
- std::string prefix = "lumina-global-index-f0-10-20-task-1-";
+ std::string prefix = "lumina-global-index-f0-10-20-";
ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_path_factory,
CreateCheckpointPathFactory(test_root, "lumina",
"f0", Range(10, 20)));
ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> path_factory,
@@ -457,13 +457,13 @@ TEST_F(LuminaGlobalIndexTest,
TestCheckpointFileManagement) {
ASSERT_EQ(second_written, 6);
ASSERT_OK(second_output->Close());
- std::string checkpoint_dir = PathUtil::JoinPath(test_root,
"index/checkpoint");
+ std::string checkpoint_dir = PathUtil::JoinPath(test_root,
"index/checkpoint/task-1");
ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(checkpoint_dir, prefix +
"9.index.ckpt"), "id-nine",
/*overwrite=*/false));
ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(checkpoint_dir, prefix +
"10.index.ckpt"), "id-ten",
/*overwrite=*/false));
std::string unrelated_file =
- PathUtil::JoinPath(checkpoint_dir,
"lumina-global-index-other-10-20-task-2-99.index.ckpt");
+ PathUtil::JoinPath(checkpoint_dir,
"lumina-global-index-other-10-20-99.index.ckpt");
ASSERT_OK(fs_->WriteFile(unrelated_file, "unrelated",
/*overwrite=*/false));
std::string malformed_file =
PathUtil::JoinPath(checkpoint_dir, prefix + "invalid-99.index.ckpt");
@@ -709,7 +709,8 @@ TEST_F(LuminaGlobalIndexTest,
TestWriteAndReadWithTagFilter) {
GlobalIndexIOMeta meta,
WriteGlobalIndex(test_root, tag_data_type, tag_options, tag_array,
Range(0, 3)));
std::vector<BasicFileStatus> checkpoint_files;
- ASSERT_OK(fs_->ListDir(PathUtil::JoinPath(test_root, "index/checkpoint"),
&checkpoint_files));
+ ASSERT_OK(
+ fs_->ListDir(PathUtil::JoinPath(test_root, "index/checkpoint/task-1"),
&checkpoint_files));
ASSERT_TRUE(checkpoint_files.empty());
ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexReader> reader,
CreateGlobalIndexReader(test_root, data_type_,
tag_options, meta));
diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp
index 97b16982..c5b42162 100644
--- a/test/inte/global_index_test.cpp
+++ b/test/inte/global_index_test.cpp
@@ -576,10 +576,12 @@ TEST_P(GlobalIndexTest,
TestWriteLuminaIndexWithCheckpoint) {
std::shared_ptr<FileSystem> checkpoint_fs = dir_->GetFileSystem();
std::string task_id = "task-1";
std::string checkpoint_dir = PathUtil::JoinPath(table_path,
"index/checkpoint");
- ASSERT_OK(checkpoint_fs->Mkdirs(checkpoint_dir));
+ std::string task_checkpoint_dir = PathUtil::JoinPath(checkpoint_dir,
task_id);
+ ASSERT_OK(checkpoint_fs->Mkdirs(task_checkpoint_dir));
+ ASSERT_OK(checkpoint_fs->Mkdirs(PathUtil::JoinPath(checkpoint_dir,
"another-task")));
auto checkpoint_path = [&](const std::string& task, int64_t id) {
- return PathUtil::JoinPath(
- checkpoint_dir,
fmt::format("lumina-global-index-f1-0-3-{}-{}.index.ckpt", task, id));
+ return PathUtil::JoinPath(PathUtil::JoinPath(checkpoint_dir, task),
+
fmt::format("lumina-global-index-f1-0-3-{}.index.ckpt", id));
};
std::string matching_checkpoint = checkpoint_path(task_id, 9);
ASSERT_OK(checkpoint_fs->WriteFile(matching_checkpoint, "invalid
checkpoint",
@@ -593,10 +595,17 @@ TEST_P(GlobalIndexTest,
TestWriteLuminaIndexWithCheckpoint) {
std::make_shared<IndexedSplitImpl>(split, std::vector<Range>{Range(0,
3)}), lumina_options,
task_id, pool_));
std::vector<BasicFileStatus> checkpoint_files;
- ASSERT_OK(checkpoint_fs->ListDir(checkpoint_dir, &checkpoint_files));
- ASSERT_EQ(checkpoint_files.size(), 1);
- ASSERT_EQ(PathUtil::GetName(checkpoint_files[0].GetPath()),
- PathUtil::GetName(unrelated_checkpoint));
+ ASSERT_OK(checkpoint_fs->ListDir(task_checkpoint_dir, &checkpoint_files));
+ ASSERT_TRUE(checkpoint_files.empty());
+ ASSERT_OK_AND_ASSIGN(bool unrelated_exists,
checkpoint_fs->Exists(unrelated_checkpoint));
+ ASSERT_TRUE(unrelated_exists);
+
+ ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(table_path,
"another-task", lumina_options));
+ ASSERT_OK_AND_ASSIGN(unrelated_exists,
checkpoint_fs->Exists(unrelated_checkpoint));
+ ASSERT_FALSE(unrelated_exists);
+ ASSERT_OK_AND_ASSIGN(bool unrelated_directory_exists,
+
checkpoint_fs->Exists(PathUtil::JoinPath(checkpoint_dir, "another-task")));
+ ASSERT_FALSE(unrelated_directory_exists);
}
TEST_P(GlobalIndexTest, TestWriteLuminaIndexWithMismatchedDimension) {
