This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new bb7cb2e5 fix(global-index): support external path for global index
checkpoints (#473)
bb7cb2e5 is described below
commit bb7cb2e5e1fe4a8a7e93c73316cda3f36ceaebe9
Author: lszskye <[email protected]>
AuthorDate: Wed Oct 28 16:50:15 2026 +0800
fix(global-index): support external path for global index checkpoints (#473)
---
docs/source/api/file_index.rst | 64 ------------
docs/source/user_guide.rst | 1 +
docs/source/{api => user_guide}/file_index.rst | 41 +-------
docs/source/user_guide/global_index.rst | 12 ++-
.../paimon/global_index/global_index_write_task.h | 8 +-
.../global_index_file_manager_test.cpp | 110 ++++++++++++++++++++-
.../core/global_index/global_index_write_task.cpp | 11 ++-
.../core/index/index_checkpoint_path_factory.h | 11 ++-
src/paimon/core/utils/file_store_path_factory.cpp | 4 +-
.../core/utils/file_store_path_factory_test.cpp | 48 ++++++---
test/inte/global_index_test.cpp | 73 ++++++++------
11 files changed, 221 insertions(+), 162 deletions(-)
diff --git a/docs/source/api/file_index.rst b/docs/source/api/file_index.rst
index b6a0eb14..c9fedf37 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<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
=========
diff --git a/docs/source/user_guide.rst b/docs/source/user_guide.rst
index c03efd88..a8d3c7f8 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/api/file_index.rst
b/docs/source/user_guide/file_index.rst
similarity index 85%
copy from docs/source/api/file_index.rst
copy to docs/source/user_guide/file_index.rst
index b6a0eb14..a402cc6e 100644
--- a/docs/source/api/file_index.rst
+++ b/docs/source/user_guide/file_index.rst
@@ -15,14 +15,11 @@
.. specific language governing permissions and limitations
.. under the License.
-===========
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,
@@ -84,37 +81,3 @@ 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
-=========
-
-.. doxygenstruct:: paimon::FullTextMatch
- :members:
-
-.. doxygenclass:: paimon::FileIndexer
- :members:
- :undoc-members:
-
-.. doxygenclass:: paimon::FileIndexerFactory
- :members:
- :undoc-members:
-
-.. doxygenclass:: paimon::FileIndexWriter
- :members:
- :undoc-members:
-
-.. doxygenclass:: paimon::FileIndexReader
- :members:
- :undoc-members:
-
-.. doxygenclass:: paimon::FileIndexFormat
- :members:
- :undoc-members:
-
-.. doxygenclass:: paimon::FileIndexResult
- :members:
- :undoc-members:
-
-.. doxygenclass:: paimon::BitmapIndexResult
- :members:
- :undoc-members:
diff --git a/docs/source/user_guide/global_index.rst
b/docs/source/user_guide/global_index.rst
index 897cfebb..c744d018 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 ``<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
------------
diff --git a/include/paimon/global_index/global_index_write_task.h
b/include/paimon/global_index/global_index_write_task.h
index 8c89a9c1..15cc6b7d 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<std::string, std::string>&
options,
const std::shared_ptr<FileSystem>&
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 d5cce4ad..f375a266 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<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());
}
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 f04d96bb..cb11fd92 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<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();
diff --git a/src/paimon/core/index/index_checkpoint_path_factory.h
b/src/paimon/core/index/index_checkpoint_path_factory.h
index 0bfa56aa..cdc043ba 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<int64_t> GetCheckpointId(const std::string&
file_name) const = 0;
- /// Returns the task's checkpoint directory after validating its
identifier.
- static Result<std::string> GetTaskDirectory(const std::string& table_path,
- const std::string& task_id) {
+ /// Returns the task's checkpoint directory under the configured global
index directory.
+ static Result<std::string> GetTaskDirectory(
+ const std::string& table_path, const std::string& task_id,
+ const std::optional<std::string>& 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 9acb6d4e..cbb767fc 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<IndexCheckpointPathFactory>(
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 447cb4f4..1d38035e 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<FileStorePathFactory> CreateFactory(const std::string&
root) const {
+ std::shared_ptr<FileStorePathFactory> CreateFactory(
+ const std::string& root,
+ const std::optional<std::string>& 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<std::string> 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<IndexCheckpointPathFactory>
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<std::string>{"", ".", "..",
" ", "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 c5b42162..a978b299 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<FileSystem> 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<IndexedSplitImpl>(split, std::vector<Range>{Range(0,
3)}), lumina_options,
- task_id, pool_));
- std::vector<BasicFileStatus> 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<std::string, std::string> 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<IndexedSplitImpl>(split,
std::vector<Range>{Range(0, 3)}),
+ checkpoint_options, task_id, pool_));
+ std::vector<BasicFileStatus> 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) {
