From e0140ac0a49afe06f083e0212c15ef05a9f8bde6 Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Tue, 27 Oct 2026 19:52:08 +0800 Subject: [PATCH 1/2] fix(global-index): support external path for global index checkpoints --- docs/source/api/file_index.rst | 64 ---------- docs/source/user_guide.rst | 1 + docs/source/user_guide/file_index.rst | 83 +++++++++++++ docs/source/user_guide/global_index.rst | 12 +- .../global_index/global_index_write_task.h | 8 +- .../global_index_file_manager_test.cpp | 111 +++++++++++++++++- .../global_index/global_index_write_task.cpp | 11 +- .../index/index_checkpoint_path_factory.h | 11 +- .../core/utils/file_store_path_factory.cpp | 4 +- .../utils/file_store_path_factory_test.cpp | 48 ++++++-- test/inte/global_index_test.cpp | 73 +++++++----- 11 files changed, 303 insertions(+), 123 deletions(-) create mode 100644 docs/source/user_guide/file_index.rst diff --git a/docs/source/api/file_index.rst b/docs/source/api/file_index.rst index b6a0eb140..c9fedf377 100644 --- a/docs/source/api/file_index.rst +++ b/docs/source/api/file_index.rst @@ -21,70 +21,6 @@ File Index .. _cpp-api-file-index: -Full-text predicates and root TopK -================================= - -``PredicateBuilder::FullTextMatch`` creates a ``LeafPredicate`` for a full-text -condition without TopK truncation. It holds a shared ``FullTextMatch`` request, -whose query uses the same JSON DSL as ``FullTextSearch``. Do not modify a request -after passing it to a predicate or read context. - -Compose full-text and scalar predicates with the existing -``PredicateBuilder::And`` and ``PredicateBuilder::Or`` factories. Every field -referenced by this tree must appear in the read schema; field indices refer to -positions in that schema. For example, the following constructs -``body matches "apple" OR price < 10``: - -.. code-block:: cpp - - auto request = std::make_shared( - "body", R"({"match":{"query":"apple"}})"); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr text, - paimon::PredicateBuilder::FullTextMatch(0, request)); - auto price = paimon::PredicateBuilder::LessThan( - 1, "price", paimon::FieldType::INT, paimon::Literal(10)); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr filter, - paimon::PredicateBuilder::Or({text, price})); - paimon::ReadContextBuilder builder(table_path); - builder.SetReadFieldNames({"body", "price"}).SetPredicate(filter); - -``SetVectorSearch`` and ``SetFullTextSearch`` configure a separate root TopK -request. They are mutually exclusive and require a positive limit. The intended -semantics are to filter by the complete predicate before ranking; the TopK search -field does not need to be projected. A full-text match inside a predicate never -has a TopK limit. - -Score output is requested through an explicit ``__INDEX_SCORE`` -projection. For example, request the score of a full-text -predicate by projecting ``_body_INDEX_SCORE``: - -.. code-block:: cpp - - auto request = std::make_shared( - "body", R"({"match":{"query":"apple"}})"); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr text, - paimon::PredicateBuilder::FullTextMatch(0, request)); - paimon::ReadContextBuilder builder(table_path); - builder.SetReadFieldNames({"body", "_body_INDEX_SCORE"}).SetPredicate(text); - -Score fields have type ``float32`` and preserve the source field's exact name, -including case. Select them with ``SetReadFieldNames`` or ``SetReadSchema``; -they have no persistent field IDs for ``SetReadFieldIds``. Reading all table -fields does not implicitly add score fields. - -Each projected score must have exactly one search condition as its source. -The source may be a root TopK request or a full-text predicate. A full-text -predicate whose score is projected cannot have an OR ancestor, and its field -cannot occur in another search condition, including the root TopK request. -An ordinary scalar predicate on the same field is allowed. For example, when -projecting the score of ``match``, ``match AND (p OR q)`` is valid, while -``match OR p`` is rejected. Without that score projection, both OR branches and -repeated full-text search fields are allowed. These checks use the effective -read projection, after resolving the read schema. - Interface ========= diff --git a/docs/source/user_guide.rst b/docs/source/user_guide.rst index c03efd88a..a8d3c7f8d 100644 --- a/docs/source/user_guide.rst +++ b/docs/source/user_guide.rst @@ -43,5 +43,6 @@ User Guide user_guide/clean user_guide/prefetch user_guide/arrow + user_guide/file_index user_guide/global_index user_guide/primary_key_global_index diff --git a/docs/source/user_guide/file_index.rst b/docs/source/user_guide/file_index.rst new file mode 100644 index 000000000..a402cc6e8 --- /dev/null +++ b/docs/source/user_guide/file_index.rst @@ -0,0 +1,83 @@ +.. 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. + +File Index +========== + +Full-text predicates and root TopK +--------------------------------- + +``PredicateBuilder::FullTextMatch`` creates a ``LeafPredicate`` for a full-text +condition without TopK truncation. It holds a shared ``FullTextMatch`` request, +whose query uses the same JSON DSL as ``FullTextSearch``. Do not modify a request +after passing it to a predicate or read context. + +Compose full-text and scalar predicates with the existing +``PredicateBuilder::And`` and ``PredicateBuilder::Or`` factories. Every field +referenced by this tree must appear in the read schema; field indices refer to +positions in that schema. For example, the following constructs +``body matches "apple" OR price < 10``: + +.. code-block:: cpp + + auto request = std::make_shared( + "body", R"({"match":{"query":"apple"}})"); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr text, + paimon::PredicateBuilder::FullTextMatch(0, request)); + auto price = paimon::PredicateBuilder::LessThan( + 1, "price", paimon::FieldType::INT, paimon::Literal(10)); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr filter, + paimon::PredicateBuilder::Or({text, price})); + paimon::ReadContextBuilder builder(table_path); + builder.SetReadFieldNames({"body", "price"}).SetPredicate(filter); + +``SetVectorSearch`` and ``SetFullTextSearch`` configure a separate root TopK +request. They are mutually exclusive and require a positive limit. The intended +semantics are to filter by the complete predicate before ranking; the TopK search +field does not need to be projected. A full-text match inside a predicate never +has a TopK limit. + +Score output is requested through an explicit ``__INDEX_SCORE`` +projection. For example, request the score of a full-text +predicate by projecting ``_body_INDEX_SCORE``: + +.. code-block:: cpp + + auto request = std::make_shared( + "body", R"({"match":{"query":"apple"}})"); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr text, + paimon::PredicateBuilder::FullTextMatch(0, request)); + paimon::ReadContextBuilder builder(table_path); + builder.SetReadFieldNames({"body", "_body_INDEX_SCORE"}).SetPredicate(text); + +Score fields have type ``float32`` and preserve the source field's exact name, +including case. Select them with ``SetReadFieldNames`` or ``SetReadSchema``; +they have no persistent field IDs for ``SetReadFieldIds``. Reading all table +fields does not implicitly add score fields. + +Each projected score must have exactly one search condition as its source. +The source may be a root TopK request or a full-text predicate. A full-text +predicate whose score is projected cannot have an OR ancestor, and its field +cannot occur in another search condition, including the root TopK request. +An ordinary scalar predicate on the same field is allowed. For example, when +projecting the score of ``match``, ``match AND (p OR q)`` is valid, while +``match OR p`` is rejected. Without that score projection, both OR branches and +repeated full-text search fields are allowed. These checks use the effective +read projection, after resolving the read schema. diff --git a/docs/source/user_guide/global_index.rst b/docs/source/user_guide/global_index.rst index 897cfebbb..c744d018c 100644 --- a/docs/source/user_guide/global_index.rst +++ b/docs/source/user_guide/global_index.rst @@ -50,9 +50,15 @@ task once it is no longer running or being retried: 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. +Checkpoints are stored in ``/index/checkpoint/`` by default. When +``global-index.external-path`` is configured, they are stored in +``/checkpoint/`` instead. + +Pass the same table path, task id, and ``global-index.external-path`` configuration used in +``WriteIndex``. Cleanup reads the configuration from the table schema and applies any supplied +options as overrides. Task ids must be unique among builds sharing the same global index directory, +including builds for different tables sharing an external path. 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 8c89a9c18..15cc6b7db 100644 --- a/include/paimon/global_index/global_index_write_task.h +++ b/include/paimon/global_index/global_index_write_task.h @@ -50,7 +50,8 @@ class PAIMON_EXPORT GlobalIndexWriteTask { /// filters). /// @param task_id When checkpoints are enabled, the caller must provide a non-empty task /// identifier that uniquely identifies an index build task within the - /// table. Reuse it when retrying the same build. If the source data, build + /// global index directory, including tables sharing an external path. + /// 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. @@ -73,9 +74,12 @@ class PAIMON_EXPORT GlobalIndexWriteTask { /// @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. + /// schema options, with these overrides taking precedence. Checkpoints + /// are deleted from the configured global index external path, or from + /// the table's index directory when no external path is configured. /// @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. + /// File system errors are propagated to the caller without being suppressed. static Status DeleteCheckpoint(const std::string& table_path, const std::string& task_id, const std::map& options, const std::shared_ptr& file_system = nullptr); 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 d5cce4adb..d4456c5f8 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 @@ -216,6 +216,61 @@ TEST_F(GlobalIndexFileManagerTest, TestCheckpointIOAndRestart) { ASSERT_TRUE(exists); } +TEST_F(GlobalIndexFileManagerTest, TestExternalCheckpointIOAndRestart) { + std::string external_path = "file:" + PathUtil::JoinPath(dir_->Str(), "external-index"); + ASSERT_OK_AND_ASSIGN( + path_factory_, + FileStorePathFactory::Create( + dir_->Str(), arrow::schema({}), /*partition_keys=*/{}, /*default_part_value=*/"", + /*identifier=*/"mock", /*data_file_prefix=*/"data-", + /*legacy_partition_name_enabled=*/true, /*external_paths=*/{}, external_path, + /*index_file_in_data_file_dir=*/false, GetDefaultPool())); + ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); + std::string directory = PathUtil::JoinPath(external_path, "checkpoint/task-1"); + std::string file_name = "lumina-global-index-vector-10-20-0.index.ckpt"; + std::string local_path = + PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1/" + file_name); + ASSERT_OK(fs_->WriteFile(local_path, "local", /*overwrite=*/false)); + ASSERT_OK_AND_ASSIGN(bool exists, manager->CheckpointExists()); + ASSERT_FALSE(exists); + ASSERT_OK_AND_ASSIGN(std::unique_ptr output, + manager->CreateCheckpointOutputStream()); + ASSERT_OK_AND_ASSIGN(int64_t written, output->Write("external", 8)); + ASSERT_EQ(written, 8); + ASSERT_OK(output->Close()); + ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(PathUtil::JoinPath(directory, file_name))); + ASSERT_TRUE(exists); + + ASSERT_OK_AND_ASSIGN(std::shared_ptr restarted, CreateManager()); + ASSERT_OK_AND_ASSIGN(exists, restarted->CheckpointExists()); + ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(std::unique_ptr input, + restarted->OpenCheckpointInputStream()); + char buffer[8]; + ASSERT_OK_AND_ASSIGN(int64_t read, input->Read(buffer, sizeof(buffer))); + ASSERT_EQ(read, 8); + ASSERT_EQ(std::string(buffer, sizeof(buffer)), "external"); + ASSERT_OK(input->Close()); + ASSERT_OK_AND_ASSIGN(std::string next_name, CreateCheckpointFile(restarted)); + ASSERT_EQ(next_name, "lumina-global-index-vector-10-20-1.index.ckpt"); + + std::vector retained_paths = { + local_path, PathUtil::JoinPath(external_path, "checkpoint/task-2/" + file_name), + PathUtil::JoinPath(directory, "lumina-global-index-other-10-20-0.index.ckpt"), + manager->ToPath("retained.index")}; + for (size_t i = 1; i < retained_paths.size(); ++i) { + ASSERT_OK(fs_->WriteFile(retained_paths[i], "retained", /*overwrite=*/false)); + } + ASSERT_OK(restarted->DeleteCheckpoint()); + ASSERT_OK_AND_ASSIGN(exists, restarted->CheckpointExists()); + ASSERT_FALSE(exists); + ASSERT_OK(restarted->DeleteCheckpoint()); + for (const std::string& path : retained_paths) { + ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(path)); + ASSERT_TRUE(exists) << path; + } +} + TEST_F(GlobalIndexFileManagerTest, TestInitializationFailureCanRetry) { ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); fs_->fail_list_ = true; @@ -357,6 +412,55 @@ TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpoint) { ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_)); } +TEST_F(GlobalIndexFileManagerTest, TestDeleteExternalTaskCheckpointFromSchemaAndOverrides) { + std::string schema_path = "file:" + PathUtil::JoinPath(dir_->Str(), "schema-index"); + std::string override_path = "file:" + PathUtil::JoinPath(dir_->Str(), "override-index"); + ASSERT_OK(CreateSchema({{Options::GLOBAL_INDEX_EXTERNAL_PATH, " " + schema_path + "/ "}})); + std::string local_directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1"); + std::string schema_directory = PathUtil::JoinPath(schema_path, "checkpoint/task-1"); + std::string override_directory = PathUtil::JoinPath(override_path, "checkpoint/task-1"); + for (const std::string& directory : {local_directory, schema_directory, override_directory}) { + ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, "nested/unfinished-checkpoint"), + "checkpoint", /*overwrite=*/false)); + } + std::vector retained_paths = { + PathUtil::JoinPath(schema_path, "checkpoint/task-2/checkpoint"), + PathUtil::JoinPath(override_path, "checkpoint/task-1-extended/checkpoint"), + PathUtil::JoinPath(schema_path, "retained.index"), + PathUtil::JoinPath(override_path, "retained.index")}; + for (const std::string& path : retained_paths) { + ASSERT_OK(fs_->WriteFile(path, "retained", /*overwrite=*/false)); + } + std::map overrides = { + {Options::GLOBAL_INDEX_EXTERNAL_PATH, " " + override_path + "/ "}}; + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", overrides, fs_)); + ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(override_directory)); + ASSERT_FALSE(exists); + ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(schema_directory)); + ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(local_directory)); + ASSERT_TRUE(exists); + ASSERT_EQ(fs_->delete_paths_, std::vector({override_directory})); + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", overrides, fs_)); + + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{})); + ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(schema_directory)); + ASSERT_FALSE(exists); + ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(local_directory)); + ASSERT_TRUE(exists); + ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{})); + 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::GLOBAL_INDEX_EXTERNAL_PATH, ""}}, fs_)); + ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(local_directory)); + ASSERT_FALSE(exists); +} + + TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointWithoutCustomFileSystem) { ASSERT_OK(CreateSchema()); ASSERT_NOK_WITH_MSG(GlobalIndexWriteTask::DeleteCheckpoint( @@ -376,12 +480,13 @@ TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointWithoutCustomFileSyst } TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointRejectsInvalidTaskId) { + ASSERT_OK(CreateSchema()); 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_NOK_WITH_MSG( + GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), invalid, /*options=*/{}, fs_), + "checkpoint task id"); } - ASSERT_EQ(fs_->list_count_, 0); ASSERT_TRUE(fs_->delete_paths_.empty()); } 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 f04d96bb0..cb11fd920 100644 --- a/src/paimon/core/global_index/global_index_write_task.cpp +++ b/src/paimon/core/global_index/global_index_write_task.cpp @@ -31,6 +31,7 @@ #include "paimon/common/utils/arrow/mem_utils.h" #include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/checked_cast.h" +#include "paimon/common/utils/path_util.h" #include "paimon/common/utils/scope_guard.h" #include "paimon/core/casting/casting_utils.h" #include "paimon/core/core_options.h" @@ -424,12 +425,16 @@ 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(); + const CoreOptions& core_options = schema_and_options.second; + PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, + core_options.CreateGlobalIndexExternalPath()); + PAIMON_ASSIGN_OR_RAISE(std::string task_directory, + IndexCheckpointPathFactory::GetTaskDirectory( + table_path, task_id, global_index_external_path)); + std::shared_ptr fs = core_options.GetFileSystem(); PAIMON_ASSIGN_OR_RAISE(bool exists, fs->Exists(task_directory)); if (!exists) { return Status::OK(); diff --git a/src/paimon/core/index/index_checkpoint_path_factory.h b/src/paimon/core/index/index_checkpoint_path_factory.h index 0bfa56aa7..cdc043bae 100644 --- a/src/paimon/core/index/index_checkpoint_path_factory.h +++ b/src/paimon/core/index/index_checkpoint_path_factory.h @@ -43,11 +43,14 @@ 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) { + /// Returns the task's checkpoint directory under the configured global index directory. + static Result GetTaskDirectory( + const std::string& table_path, const std::string& task_id, + const std::optional& global_index_external_path) { PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint task id", task_id)); - return PathUtil::JoinPath(PathUtil::JoinPath(table_path, "index/checkpoint"), task_id); + std::string index_directory = + global_index_external_path.value_or(PathUtil::JoinPath(table_path, "index")); + return PathUtil::JoinPath(PathUtil::JoinPath(index_directory, "checkpoint"), task_id); } }; diff --git a/src/paimon/core/utils/file_store_path_factory.cpp b/src/paimon/core/utils/file_store_path_factory.cpp index 9acb6d4e3..cbb767fcd 100644 --- a/src/paimon/core/utils/file_store_path_factory.cpp +++ b/src/paimon/core/utils/file_store_path_factory.cpp @@ -248,8 +248,8 @@ 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_ASSIGN_OR_RAISE(std::string directory, - IndexCheckpointPathFactory::GetTaskDirectory(root_, task_id)); + PAIMON_ASSIGN_OR_RAISE(std::string directory, IndexCheckpointPathFactory::GetTaskDirectory( + root_, task_id, global_index_external_path_)); std::string file_name_prefix = fmt::format("{}-global-index-{}-{}-{}-", index_type, field_name, range.from, range.to); return std::unique_ptr( 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 447cb4f47..1d38035e3 100644 --- a/src/paimon/core/utils/file_store_path_factory_test.cpp +++ b/src/paimon/core/utils/file_store_path_factory_test.cpp @@ -53,7 +53,9 @@ class FileStorePathFactoryTest : public ::testing::Test { } void TearDown() override {} - std::shared_ptr CreateFactory(const std::string& root) const { + std::shared_ptr CreateFactory( + const std::string& root, + const std::optional& global_index_external_path = std::nullopt) const { arrow::FieldVector fields = {arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), arrow::field("f2", arrow::int8()), @@ -76,13 +78,13 @@ class FileStorePathFactoryTest : public ::testing::Test { EXPECT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap(raw_options)); EXPECT_OK_AND_ASSIGN(std::vector external_paths, options.CreateExternalPaths()); - EXPECT_OK_AND_ASSIGN(auto path_factory, - FileStorePathFactory::Create( - root, schema, {"f0", "f3"}, options.GetPartitionDefaultName(), - options.GetFileFormat()->Identifier(), options.DataFilePrefix(), - options.LegacyPartitionNameEnabled(), external_paths, - /*global_index_external_path=*/std::nullopt, - options.IndexFileInDataFileDir(), mem_pool_)); + EXPECT_OK_AND_ASSIGN( + auto path_factory, + FileStorePathFactory::Create( + root, schema, {"f0", "f3"}, options.GetPartitionDefaultName(), + options.GetFileFormat()->Identifier(), options.DataFilePrefix(), + options.LegacyPartitionNameEnabled(), external_paths, global_index_external_path, + options.IndexFileInDataFileDir(), mem_pool_)); return path_factory; } @@ -601,8 +603,8 @@ TEST_F(FileStorePathFactoryTest, TestCreateGlobalIndexCheckpointPathFactory) { file_store_path_factory->CreateGlobalIndexCheckpointPathFactory( "lumina", "vector", Range(10, 20), "task-1")); 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_OK_AND_ASSIGN(std::string task_directory, IndexCheckpointPathFactory::GetTaskDirectory( + dir->Str(), "task-1", std::nullopt)); ASSERT_EQ(task_directory, checkpoint_dir); ASSERT_EQ(checkpoint_path_factory->GetDirectoryPath(), checkpoint_dir); std::string first_path = checkpoint_path_factory->NewPath(0); @@ -634,6 +636,30 @@ TEST_F(FileStorePathFactoryTest, TestCreateGlobalIndexCheckpointPathFactory) { ASSERT_FALSE(exists); } +TEST_F(FileStorePathFactoryTest, TestCreateGlobalIndexCheckpointPathFactoryWithExternalPath) { + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + for (const std::string& external_path : + {"file://" + dir->Str() + "/external-index/", std::string("s3://bucket/external-index")}) { + auto factory = CreateFactory(dir->Str(), external_path); + ASSERT_OK_AND_ASSIGN(std::unique_ptr checkpoint_factory, + factory->CreateGlobalIndexCheckpointPathFactory( + "lumina", "vector", Range(10, 20), "task-1")); + std::string checkpoint_dir = PathUtil::JoinPath(external_path, "checkpoint/task-1"); + ASSERT_OK_AND_ASSIGN( + std::string task_directory, + IndexCheckpointPathFactory::GetTaskDirectory(dir->Str(), "task-1", external_path)); + ASSERT_EQ(task_directory, checkpoint_dir); + ASSERT_EQ(checkpoint_factory->GetDirectoryPath(), checkpoint_dir); + std::string file_name = "lumina-global-index-vector-10-20-9.index.ckpt"; + ASSERT_EQ(checkpoint_factory->NewPath(9), PathUtil::JoinPath(checkpoint_dir, file_name)); + ASSERT_EQ(checkpoint_factory->ToPath(file_name), checkpoint_factory->NewPath(9)); + } + ASSERT_OK_AND_ASSIGN(bool exists, dir->GetFileSystem()->Exists( + PathUtil::JoinPath(dir->Str(), "external-index"))); + ASSERT_FALSE(exists); +} + TEST_F(FileStorePathFactoryTest, TestCheckpointFileIdPath) { auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); @@ -707,7 +733,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(IndexCheckpointPathFactory::GetTaskDirectory(dir->Str(), invalid, std::nullopt)); ASSERT_NOK(factory->CreateGlobalIndexCheckpointPathFactory(invalid, "vector", Range(10, 20), "task")); ASSERT_NOK(factory->CreateGlobalIndexCheckpointPathFactory("lumina", invalid, Range(10, 20), diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index c5b421629..a978b2999 100644 --- a/test/inte/global_index_test.cpp +++ b/test/inte/global_index_test.cpp @@ -575,37 +575,48 @@ TEST_P(GlobalIndexTest, TestWriteLuminaIndexWithCheckpoint) { ASSERT_OK_AND_ASSIGN(auto split, ScanData(table_path, /*partition_filters=*/{})); std::shared_ptr checkpoint_fs = dir_->GetFileSystem(); std::string task_id = "task-1"; - std::string checkpoint_dir = PathUtil::JoinPath(table_path, "index/checkpoint"); - 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(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", - /*overwrite=*/false)); - std::string unrelated_checkpoint = checkpoint_path("another-task", 999); - ASSERT_OK(checkpoint_fs->WriteFile(unrelated_checkpoint, "another task", - /*overwrite=*/false)); - - ASSERT_OK(GlobalIndexWriteTask::WriteIndex( - table_path, "f1", "lumina", - std::make_shared(split, std::vector{Range(0, 3)}), lumina_options, - task_id, pool_)); - std::vector checkpoint_files; - 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); + auto external_dir = UniqueTestDirectory::Create("local"); + ASSERT_TRUE(external_dir); + for (bool use_external_path : {false, true}) { + std::map checkpoint_options = lumina_options; + std::string checkpoint_dir = PathUtil::JoinPath(table_path, "index/checkpoint"); + if (use_external_path) { + checkpoint_options["global-index.external-path"] = "FILE://" + external_dir->Str(); + checkpoint_dir = PathUtil::JoinPath(external_dir->Str(), "checkpoint"); + } + 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(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", + /*overwrite=*/false)); + std::string unrelated_checkpoint = checkpoint_path("another-task", 999); + ASSERT_OK(checkpoint_fs->WriteFile(unrelated_checkpoint, "another task", + /*overwrite=*/false)); + + ASSERT_OK(GlobalIndexWriteTask::WriteIndex( + table_path, "f1", "lumina", + std::make_shared(split, std::vector{Range(0, 3)}), + checkpoint_options, task_id, pool_)); + std::vector checkpoint_files; + 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", checkpoint_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) { From 1752d3da2aa5ba2b981cc7f0aced0bd02af12d79 Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Wed, 28 Oct 2026 15:11:44 +0800 Subject: [PATCH 2/2] fix pre-commit --- src/paimon/core/global_index/global_index_file_manager_test.cpp | 1 - 1 file changed, 1 deletion(-) 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 d4456c5f8..f375a266e 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 @@ -460,7 +460,6 @@ TEST_F(GlobalIndexFileManagerTest, TestDeleteExternalTaskCheckpointFromSchemaAnd ASSERT_FALSE(exists); } - TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointWithoutCustomFileSystem) { ASSERT_OK(CreateSchema()); ASSERT_NOK_WITH_MSG(GlobalIndexWriteTask::DeleteCheckpoint(