Skip to content

Commit bb7cb2e

Browse files
lszskyelszskye
authored
fix(global-index): support external path for global index checkpoints (#473)
1 parent 1b3c690 commit bb7cb2e

11 files changed

Lines changed: 302 additions & 123 deletions

File tree

‎docs/source/api/file_index.rst‎

Lines changed: 0 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -21,70 +21,6 @@ File Index
2121

2222
.. _cpp-api-file-index:
2323

24-
Full-text predicates and root TopK
25-
=================================
26-
27-
``PredicateBuilder::FullTextMatch`` creates a ``LeafPredicate`` for a full-text
28-
condition without TopK truncation. It holds a shared ``FullTextMatch`` request,
29-
whose query uses the same JSON DSL as ``FullTextSearch``. Do not modify a request
30-
after passing it to a predicate or read context.
31-
32-
Compose full-text and scalar predicates with the existing
33-
``PredicateBuilder::And`` and ``PredicateBuilder::Or`` factories. Every field
34-
referenced by this tree must appear in the read schema; field indices refer to
35-
positions in that schema. For example, the following constructs
36-
``body matches "apple" OR price < 10``:
37-
38-
.. code-block:: cpp
39-
40-
auto request = std::make_shared<paimon::FullTextMatch>(
41-
"body", R"({"match":{"query":"apple"}})");
42-
PAIMON_ASSIGN_OR_RAISE(
43-
std::shared_ptr<paimon::Predicate> text,
44-
paimon::PredicateBuilder::FullTextMatch(0, request));
45-
auto price = paimon::PredicateBuilder::LessThan(
46-
1, "price", paimon::FieldType::INT, paimon::Literal(10));
47-
PAIMON_ASSIGN_OR_RAISE(
48-
std::shared_ptr<paimon::Predicate> filter,
49-
paimon::PredicateBuilder::Or({text, price}));
50-
paimon::ReadContextBuilder builder(table_path);
51-
builder.SetReadFieldNames({"body", "price"}).SetPredicate(filter);
52-
53-
``SetVectorSearch`` and ``SetFullTextSearch`` configure a separate root TopK
54-
request. They are mutually exclusive and require a positive limit. The intended
55-
semantics are to filter by the complete predicate before ranking; the TopK search
56-
field does not need to be projected. A full-text match inside a predicate never
57-
has a TopK limit.
58-
59-
Score output is requested through an explicit ``_<field>_INDEX_SCORE``
60-
projection. For example, request the score of a full-text
61-
predicate by projecting ``_body_INDEX_SCORE``:
62-
63-
.. code-block:: cpp
64-
65-
auto request = std::make_shared<paimon::FullTextMatch>(
66-
"body", R"({"match":{"query":"apple"}})");
67-
PAIMON_ASSIGN_OR_RAISE(
68-
std::shared_ptr<paimon::Predicate> text,
69-
paimon::PredicateBuilder::FullTextMatch(0, request));
70-
paimon::ReadContextBuilder builder(table_path);
71-
builder.SetReadFieldNames({"body", "_body_INDEX_SCORE"}).SetPredicate(text);
72-
73-
Score fields have type ``float32`` and preserve the source field's exact name,
74-
including case. Select them with ``SetReadFieldNames`` or ``SetReadSchema``;
75-
they have no persistent field IDs for ``SetReadFieldIds``. Reading all table
76-
fields does not implicitly add score fields.
77-
78-
Each projected score must have exactly one search condition as its source.
79-
The source may be a root TopK request or a full-text predicate. A full-text
80-
predicate whose score is projected cannot have an OR ancestor, and its field
81-
cannot occur in another search condition, including the root TopK request.
82-
An ordinary scalar predicate on the same field is allowed. For example, when
83-
projecting the score of ``match``, ``match AND (p OR q)`` is valid, while
84-
``match OR p`` is rejected. Without that score projection, both OR branches and
85-
repeated full-text search fields are allowed. These checks use the effective
86-
read projection, after resolving the read schema.
87-
8824
Interface
8925
=========
9026

‎docs/source/user_guide.rst‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,5 +43,6 @@ User Guide
4343
user_guide/clean
4444
user_guide/prefetch
4545
user_guide/arrow
46+
user_guide/file_index
4647
user_guide/global_index
4748
user_guide/primary_key_global_index

‎docs/source/user_guide/file_index.rst‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
.. Licensed to the Apache Software Foundation (ASF) under one
2+
.. or more contributor license agreements. See the NOTICE file
3+
.. distributed with this work for additional information
4+
.. regarding copyright ownership. The ASF licenses this file
5+
.. to you under the Apache License, Version 2.0 (the
6+
.. "License"); you may not use this file except in compliance
7+
.. with the License. You may obtain a copy of the License at
8+
9+
.. http://www.apache.org/licenses/LICENSE-2.0
10+
11+
.. Unless required by applicable law or agreed to in writing,
12+
.. software distributed under the License is distributed on an
13+
.. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
.. KIND, either express or implied. See the License for the
15+
.. specific language governing permissions and limitations
16+
.. under the License.
17+
18+
File Index
19+
==========
20+
21+
Full-text predicates and root TopK
22+
---------------------------------
23+
24+
``PredicateBuilder::FullTextMatch`` creates a ``LeafPredicate`` for a full-text
25+
condition without TopK truncation. It holds a shared ``FullTextMatch`` request,
26+
whose query uses the same JSON DSL as ``FullTextSearch``. Do not modify a request
27+
after passing it to a predicate or read context.
28+
29+
Compose full-text and scalar predicates with the existing
30+
``PredicateBuilder::And`` and ``PredicateBuilder::Or`` factories. Every field
31+
referenced by this tree must appear in the read schema; field indices refer to
32+
positions in that schema. For example, the following constructs
33+
``body matches "apple" OR price < 10``:
34+
35+
.. code-block:: cpp
36+
37+
auto request = std::make_shared<paimon::FullTextMatch>(
38+
"body", R"({"match":{"query":"apple"}})");
39+
PAIMON_ASSIGN_OR_RAISE(
40+
std::shared_ptr<paimon::Predicate> text,
41+
paimon::PredicateBuilder::FullTextMatch(0, request));
42+
auto price = paimon::PredicateBuilder::LessThan(
43+
1, "price", paimon::FieldType::INT, paimon::Literal(10));
44+
PAIMON_ASSIGN_OR_RAISE(
45+
std::shared_ptr<paimon::Predicate> filter,
46+
paimon::PredicateBuilder::Or({text, price}));
47+
paimon::ReadContextBuilder builder(table_path);
48+
builder.SetReadFieldNames({"body", "price"}).SetPredicate(filter);
49+
50+
``SetVectorSearch`` and ``SetFullTextSearch`` configure a separate root TopK
51+
request. They are mutually exclusive and require a positive limit. The intended
52+
semantics are to filter by the complete predicate before ranking; the TopK search
53+
field does not need to be projected. A full-text match inside a predicate never
54+
has a TopK limit.
55+
56+
Score output is requested through an explicit ``_<field>_INDEX_SCORE``
57+
projection. For example, request the score of a full-text
58+
predicate by projecting ``_body_INDEX_SCORE``:
59+
60+
.. code-block:: cpp
61+
62+
auto request = std::make_shared<paimon::FullTextMatch>(
63+
"body", R"({"match":{"query":"apple"}})");
64+
PAIMON_ASSIGN_OR_RAISE(
65+
std::shared_ptr<paimon::Predicate> text,
66+
paimon::PredicateBuilder::FullTextMatch(0, request));
67+
paimon::ReadContextBuilder builder(table_path);
68+
builder.SetReadFieldNames({"body", "_body_INDEX_SCORE"}).SetPredicate(text);
69+
70+
Score fields have type ``float32`` and preserve the source field's exact name,
71+
including case. Select them with ``SetReadFieldNames`` or ``SetReadSchema``;
72+
they have no persistent field IDs for ``SetReadFieldIds``. Reading all table
73+
fields does not implicitly add score fields.
74+
75+
Each projected score must have exactly one search condition as its source.
76+
The source may be a root TopK request or a full-text predicate. A full-text
77+
predicate whose score is projected cannot have an OR ancestor, and its field
78+
cannot occur in another search condition, including the root TopK request.
79+
An ordinary scalar predicate on the same field is allowed. For example, when
80+
projecting the score of ``match``, ``match AND (p OR q)`` is valid, while
81+
``match OR p`` is rejected. Without that score projection, both OR branches and
82+
repeated full-text search fields are allowed. These checks use the effective
83+
read projection, after resolving the read schema.

‎docs/source/user_guide/global_index.rst‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -50,9 +50,15 @@ task once it is no longer running or being retried:
5050
PAIMON_RETURN_NOT_OK(paimon::GlobalIndexWriteTask::DeleteCheckpoint(
5151
table_path, "build-task-1", options, file_system));
5252
53-
Pass the same table path and task id used in ``WriteIndex``. Task ids must be unique within the
54-
table. The cleanup recursively removes the whole task directory, other tasks' directories and completed
55-
index files are preserved.
53+
Checkpoints are stored in ``<table_path>/index/checkpoint/<task_id>`` by default. When
54+
``global-index.external-path`` is configured, they are stored in
55+
``<global-index.external-path>/checkpoint/<task_id>`` instead.
56+
57+
Pass the same table path, task id, and ``global-index.external-path`` configuration used in
58+
``WriteIndex``. Cleanup reads the configuration from the table schema and applies any supplied
59+
options as overrides. Task ids must be unique among builds sharing the same global index directory,
60+
including builds for different tables sharing an external path. The cleanup recursively removes
61+
the whole task directory, other tasks' directories and completed index files are preserved.
5662

5763
Bitmap Index
5864
------------

‎include/paimon/global_index/global_index_write_task.h‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,8 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
5050
/// filters).
5151
/// @param task_id When checkpoints are enabled, the caller must provide a non-empty task
5252
/// identifier that uniquely identifies an index build task within the
53-
/// table. Reuse it when retrying the same build. If the source data, build
53+
/// global index directory, including tables sharing an external path.
54+
/// Reuse it when retrying the same build. If the source data, build
5455
/// configuration, or build source code changes, use a new identifier;
5556
/// otherwise, the index build may fail. Pass nullopt when checkpoints are
5657
/// disabled. Index types without checkpoint support ignore this value.
@@ -73,9 +74,12 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
7374
/// @param table_path Path to the table root directory used for the build.
7475
/// @param task_id Non-empty, unique task identifier passed to `WriteIndex`.
7576
/// @param options Configuration overrides. Used to load the schema, then merged with
76-
/// schema options, with these overrides taking precedence.
77+
/// schema options, with these overrides taking precedence. Checkpoints
78+
/// are deleted from the configured global index external path, or from
79+
/// the table's index directory when no external path is configured.
7780
/// @param file_system Specifies the file system. If `nullptr`, use default file system.
7881
/// @return OK if the task directory was deleted or does not exist.
82+
/// File system errors are propagated to the caller without being suppressed.
7983
static Status DeleteCheckpoint(const std::string& table_path, const std::string& task_id,
8084
const std::map<std::string, std::string>& options,
8185
const std::shared_ptr<FileSystem>& file_system = nullptr);

‎src/paimon/core/global_index/global_index_file_manager_test.cpp‎

Lines changed: 107 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -216,6 +216,61 @@ TEST_F(GlobalIndexFileManagerTest, TestCheckpointIOAndRestart) {
216216
ASSERT_TRUE(exists);
217217
}
218218

219+
TEST_F(GlobalIndexFileManagerTest, TestExternalCheckpointIOAndRestart) {
220+
std::string external_path = "file:" + PathUtil::JoinPath(dir_->Str(), "external-index");
221+
ASSERT_OK_AND_ASSIGN(
222+
path_factory_,
223+
FileStorePathFactory::Create(
224+
dir_->Str(), arrow::schema({}), /*partition_keys=*/{}, /*default_part_value=*/"",
225+
/*identifier=*/"mock", /*data_file_prefix=*/"data-",
226+
/*legacy_partition_name_enabled=*/true, /*external_paths=*/{}, external_path,
227+
/*index_file_in_data_file_dir=*/false, GetDefaultPool()));
228+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager, CreateManager());
229+
std::string directory = PathUtil::JoinPath(external_path, "checkpoint/task-1");
230+
std::string file_name = "lumina-global-index-vector-10-20-0.index.ckpt";
231+
std::string local_path =
232+
PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1/" + file_name);
233+
ASSERT_OK(fs_->WriteFile(local_path, "local", /*overwrite=*/false));
234+
ASSERT_OK_AND_ASSIGN(bool exists, manager->CheckpointExists());
235+
ASSERT_FALSE(exists);
236+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> output,
237+
manager->CreateCheckpointOutputStream());
238+
ASSERT_OK_AND_ASSIGN(int64_t written, output->Write("external", 8));
239+
ASSERT_EQ(written, 8);
240+
ASSERT_OK(output->Close());
241+
ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(PathUtil::JoinPath(directory, file_name)));
242+
ASSERT_TRUE(exists);
243+
244+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> restarted, CreateManager());
245+
ASSERT_OK_AND_ASSIGN(exists, restarted->CheckpointExists());
246+
ASSERT_TRUE(exists);
247+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input,
248+
restarted->OpenCheckpointInputStream());
249+
char buffer[8];
250+
ASSERT_OK_AND_ASSIGN(int64_t read, input->Read(buffer, sizeof(buffer)));
251+
ASSERT_EQ(read, 8);
252+
ASSERT_EQ(std::string(buffer, sizeof(buffer)), "external");
253+
ASSERT_OK(input->Close());
254+
ASSERT_OK_AND_ASSIGN(std::string next_name, CreateCheckpointFile(restarted));
255+
ASSERT_EQ(next_name, "lumina-global-index-vector-10-20-1.index.ckpt");
256+
257+
std::vector<std::string> retained_paths = {
258+
local_path, PathUtil::JoinPath(external_path, "checkpoint/task-2/" + file_name),
259+
PathUtil::JoinPath(directory, "lumina-global-index-other-10-20-0.index.ckpt"),
260+
manager->ToPath("retained.index")};
261+
for (size_t i = 1; i < retained_paths.size(); ++i) {
262+
ASSERT_OK(fs_->WriteFile(retained_paths[i], "retained", /*overwrite=*/false));
263+
}
264+
ASSERT_OK(restarted->DeleteCheckpoint());
265+
ASSERT_OK_AND_ASSIGN(exists, restarted->CheckpointExists());
266+
ASSERT_FALSE(exists);
267+
ASSERT_OK(restarted->DeleteCheckpoint());
268+
for (const std::string& path : retained_paths) {
269+
ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(path));
270+
ASSERT_TRUE(exists) << path;
271+
}
272+
}
273+
219274
TEST_F(GlobalIndexFileManagerTest, TestInitializationFailureCanRetry) {
220275
ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager, CreateManager());
221276
fs_->fail_list_ = true;
@@ -357,6 +412,54 @@ TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpoint) {
357412
ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}, fs_));
358413
}
359414

415+
TEST_F(GlobalIndexFileManagerTest, TestDeleteExternalTaskCheckpointFromSchemaAndOverrides) {
416+
std::string schema_path = "file:" + PathUtil::JoinPath(dir_->Str(), "schema-index");
417+
std::string override_path = "file:" + PathUtil::JoinPath(dir_->Str(), "override-index");
418+
ASSERT_OK(CreateSchema({{Options::GLOBAL_INDEX_EXTERNAL_PATH, " " + schema_path + "/ "}}));
419+
std::string local_directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/task-1");
420+
std::string schema_directory = PathUtil::JoinPath(schema_path, "checkpoint/task-1");
421+
std::string override_directory = PathUtil::JoinPath(override_path, "checkpoint/task-1");
422+
for (const std::string& directory : {local_directory, schema_directory, override_directory}) {
423+
ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, "nested/unfinished-checkpoint"),
424+
"checkpoint", /*overwrite=*/false));
425+
}
426+
std::vector<std::string> retained_paths = {
427+
PathUtil::JoinPath(schema_path, "checkpoint/task-2/checkpoint"),
428+
PathUtil::JoinPath(override_path, "checkpoint/task-1-extended/checkpoint"),
429+
PathUtil::JoinPath(schema_path, "retained.index"),
430+
PathUtil::JoinPath(override_path, "retained.index")};
431+
for (const std::string& path : retained_paths) {
432+
ASSERT_OK(fs_->WriteFile(path, "retained", /*overwrite=*/false));
433+
}
434+
std::map<std::string, std::string> overrides = {
435+
{Options::GLOBAL_INDEX_EXTERNAL_PATH, " " + override_path + "/ "}};
436+
ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", overrides, fs_));
437+
ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(override_directory));
438+
ASSERT_FALSE(exists);
439+
ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(schema_directory));
440+
ASSERT_TRUE(exists);
441+
ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(local_directory));
442+
ASSERT_TRUE(exists);
443+
ASSERT_EQ(fs_->delete_paths_, std::vector<std::string>({override_directory}));
444+
ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", overrides, fs_));
445+
446+
ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}));
447+
ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(schema_directory));
448+
ASSERT_FALSE(exists);
449+
ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(local_directory));
450+
ASSERT_TRUE(exists);
451+
ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), "task-1", /*options=*/{}));
452+
for (const std::string& path : retained_paths) {
453+
ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(path));
454+
ASSERT_TRUE(exists) << path;
455+
}
456+
457+
ASSERT_OK(GlobalIndexWriteTask::DeleteCheckpoint(
458+
dir_->Str(), "task-1", {{Options::GLOBAL_INDEX_EXTERNAL_PATH, ""}}, fs_));
459+
ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(local_directory));
460+
ASSERT_FALSE(exists);
461+
}
462+
360463
TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointWithoutCustomFileSystem) {
361464
ASSERT_OK(CreateSchema());
362465
ASSERT_NOK_WITH_MSG(GlobalIndexWriteTask::DeleteCheckpoint(
@@ -376,12 +479,13 @@ TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointWithoutCustomFileSyst
376479
}
377480

378481
TEST_F(GlobalIndexFileManagerTest, TestDeleteTaskCheckpointRejectsInvalidTaskId) {
482+
ASSERT_OK(CreateSchema());
379483
for (const std::string& invalid : std::vector<std::string>{"", ".", "..", " ", "a/b", "a\\b",
380484
"a\nb", std::string("a\0b", 3)}) {
381-
ASSERT_NOK(
382-
GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), invalid, /*options=*/{}, fs_));
485+
ASSERT_NOK_WITH_MSG(
486+
GlobalIndexWriteTask::DeleteCheckpoint(dir_->Str(), invalid, /*options=*/{}, fs_),
487+
"checkpoint task id");
383488
}
384-
ASSERT_EQ(fs_->list_count_, 0);
385489
ASSERT_TRUE(fs_->delete_paths_.empty());
386490
}
387491

‎src/paimon/core/global_index/global_index_write_task.cpp‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
#include "paimon/common/utils/arrow/mem_utils.h"
3232
#include "paimon/common/utils/arrow/status_utils.h"
3333
#include "paimon/common/utils/checked_cast.h"
34+
#include "paimon/common/utils/path_util.h"
3435
#include "paimon/common/utils/scope_guard.h"
3536
#include "paimon/core/casting/casting_utils.h"
3637
#include "paimon/core/core_options.h"
@@ -424,12 +425,16 @@ Status GlobalIndexWriteTask::DeleteCheckpoint(const std::string& table_path,
424425
const std::string& task_id,
425426
const std::map<std::string, std::string>& options,
426427
const std::shared_ptr<FileSystem>& file_system) {
427-
PAIMON_ASSIGN_OR_RAISE(std::string task_directory,
428-
IndexCheckpointPathFactory::GetTaskDirectory(table_path, task_id));
429428
std::pair<std::shared_ptr<TableSchema>, CoreOptions> schema_and_options;
430429
PAIMON_ASSIGN_OR_RAISE(schema_and_options,
431430
LoadSchemaAndOptions(table_path, options, file_system));
432-
std::shared_ptr<FileSystem> fs = schema_and_options.second.GetFileSystem();
431+
const CoreOptions& core_options = schema_and_options.second;
432+
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> global_index_external_path,
433+
core_options.CreateGlobalIndexExternalPath());
434+
PAIMON_ASSIGN_OR_RAISE(std::string task_directory,
435+
IndexCheckpointPathFactory::GetTaskDirectory(
436+
table_path, task_id, global_index_external_path));
437+
std::shared_ptr<FileSystem> fs = core_options.GetFileSystem();
433438
PAIMON_ASSIGN_OR_RAISE(bool exists, fs->Exists(task_directory));
434439
if (!exists) {
435440
return Status::OK();

0 commit comments

Comments
 (0)