fix(global-index): support external path for global index checkpoints

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<paimon::FullTextMatch>(
-       "body", R"({"match":{"query":"apple"}})");
-   PAIMON_ASSIGN_OR_RAISE(
-       std::shared_ptr<paimon::Predicate> 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<paimon::Predicate> 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 ``_<field>_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<paimon::FullTextMatch>(
-       "body", R"({"match":{"query":"apple"}})");
-   PAIMON_ASSIGN_OR_RAISE(
-       std::shared_ptr<paimon::Predicate> 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
 =========
 

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

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<paimon::FullTextMatch>(
+       "body", R"({"match":{"query":"apple"}})");
+   PAIMON_ASSIGN_OR_RAISE(
+       std::shared_ptr<paimon::Predicate> 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<paimon::Predicate> 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 ``_<field>_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<paimon::FullTextMatch>(
+       "body", R"({"match":{"query":"apple"}})");
+   PAIMON_ASSIGN_OR_RAISE(
+       std::shared_ptr<paimon::Predicate> 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.

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 ``<table_path>/index/checkpoint/<task_id>`` by default. When
+``global-index.external-path`` is configured, they are stored in
+``<global-index.external-path>/checkpoint/<task_id>`` 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
 ------------

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<std::string, std::string>& options,
                                    const std::shared_ptr<FileSystem>& file_system = nullptr);

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<GlobalIndexFileManager> 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<OutputStream> 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<GlobalIndexFileManager> restarted, CreateManager());
+    ASSERT_OK_AND_ASSIGN(exists, restarted->CheckpointExists());
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> 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<std::string> 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<GlobalIndexFileManager> manager, CreateManager());
     fs_->fail_list_ = true;
@@ -357,6 +412,54 @@ 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<std::string> 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<std::string, std::string> 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<std::string>({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 +479,13 @@ TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointWithoutCustomFileSyst
 }
 
 TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointRejectsInvalidTaskId) {
+    ASSERT_OK(CreateSchema());
     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_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());
 }
 

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<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();
+    const CoreOptions& core_options = schema_and_options.second;
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> 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<FileSystem> fs = core_options.GetFileSystem();
     PAIMON_ASSIGN_OR_RAISE(bool exists, fs->Exists(task_directory));
     if (!exists) {
         return Status::OK();