From 8fc88447876de13d010b3ef3966b15d0a8728138 Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Mon, 26 Oct 2026 16:32:09 +0800 Subject: [PATCH] feat(global-index): add checkpoint cleanup by task id --- docs/source/user_guide/global_index.rst | 16 ++ .../global_index/global_index_write_task.h | 20 ++- .../global_index_file_manager_test.cpp | 149 ++++++++++++++++-- .../global_index/global_index_write_task.cpp | 60 +++++-- .../index/index_checkpoint_path_factory.h | 11 ++ .../core/utils/file_store_path_factory.cpp | 8 +- .../utils/file_store_path_factory_test.cpp | 24 ++- .../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 f3dca7445..897cfebbb 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 391c02caf..8c89a9c18 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& options, const std::optional& task_id, const std::shared_ptr& pool, const std::shared_ptr& 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& options, + const std::shared_ptr& 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 ab27ad2e0..d5cce4adb 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 +#include #include #include @@ -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 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 delete_paths_; }; void SetUp() override { @@ -80,6 +100,17 @@ class GlobalIndexFileManagerTest : public ::testing::Test { return PathUtil::GetName(uri); } + Status CreateSchema(const std::map& options = {}) const { + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr 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 fs_ = std::make_shared(); std::unique_ptr dir_; std::shared_ptr path_factory_; @@ -140,8 +171,8 @@ TEST_F(GlobalIndexFileManagerTest, TestCheckpointNotConfigured) { TEST_F(GlobalIndexFileManagerTest, TestCheckpointIOAndRestart) { ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 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 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 input, manager->OpenCheckpointInputStream()); @@ -243,8 +274,8 @@ TEST_F(GlobalIndexFileManagerTest, TestReadsDoNotInitializeFileId) { TEST_F(GlobalIndexFileManagerTest, TestFileIdOverflow) { ASSERT_OK_AND_ASSIGN(std::shared_ptr 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::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 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 retained_tasks = {"task-2", "task-1-extended", "task-1-9", "task-1-"}; + std::vector 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({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 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{"", ".", "..", " ", "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 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({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 d5bd13e99..f04d96bb0 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 +#include #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, CoreOptions>> LoadSchemaAndOptions( + const std::string& table_path, const std::map& options, + const std::shared_ptr& 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> 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> CreateGlobalIndexer(const std::string& index_type, const CoreOptions& core_options) { PAIMON_ASSIGN_OR_RAISE(std::unique_ptr indexer, @@ -342,22 +364,10 @@ Result> GlobalIndexWriteTask::WriteIndex( std::shared_ptr pool = memory_pool ? memory_pool : GetDefaultPool(); std::shared_ptr 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> 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, 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 indexer, CreateGlobalIndexer(index_type, core_options)); PAIMON_ASSIGN_OR_RAISE(std::optional> extra_field_names, @@ -410,4 +420,22 @@ Result> GlobalIndexWriteTask::WriteIndex( extra_field_ids); } +Status GlobalIndexWriteTask::DeleteCheckpoint(const std::string& table_path, + const std::string& task_id, + const std::map& options, + const std::shared_ptr& file_system) { + PAIMON_ASSIGN_OR_RAISE(std::string task_directory, + IndexCheckpointPathFactory::GetTaskDirectory(table_path, task_id)); + std::pair, CoreOptions> schema_and_options; + PAIMON_ASSIGN_OR_RAISE(schema_and_options, + LoadSchemaAndOptions(table_path, options, file_system)); + std::shared_ptr 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 0c1b13e51..0bfa56aa7 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 #include +#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 GetCheckpointId(const std::string& file_name) const = 0; + + /// Returns the task's checkpoint directory after validating its identifier. + static Result 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 ac15ad638..9acb6d4e3 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( std::make_unique(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 bfb2b9621..447cb4f47 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 file_store_path_factory = CreateFactory(dir->Str()); std::shared_ptr 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 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 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{"", ".", "..", " ", "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 e3451c44b..0d5605f49 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 checkpoint_path_factory, CreateCheckpointPathFactory(test_root, "lumina", "f0", Range(10, 20))); ASSERT_OK_AND_ASSIGN(std::shared_ptr 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 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 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 97b169823..c5b421629 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 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(split, std::vector{Range(0, 3)}), lumina_options, task_id, pool_)); std::vector 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) {