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 9509811d feat(global-index): support checkpoints for Lumina index
builds (#382)
9509811d is described below
commit 9509811dc53a6cbecb35f59a06f43a5263391442
Author: lszskye <[email protected]>
AuthorDate: Wed Sep 23 17:45:42 2026 -0700
feat(global-index): support checkpoints for Lumina index builds (#382)
---
.../paimon/global_index/global_index_write_task.h | 12 +-
include/paimon/global_index/global_indexer.h | 5 +
.../io/global_index_checkpoint_file_manager.h | 53 +++
src/paimon/CMakeLists.txt | 1 +
.../bitmap/bitmap_global_index_test.cpp | 6 +-
.../global_index/global_indexer_factory_test.cpp | 1 +
.../rangebitmap/range_bitmap_global_index_test.cpp | 6 +-
.../core/global_index/global_index_file_manager.h | 123 +++++-
.../global_index_file_manager_test.cpp | 294 ++++++++++++++
.../core/global_index/global_index_scan_impl.cpp | 4 +-
.../core/global_index/global_index_write_task.cpp | 37 +-
.../core/index/index_checkpoint_path_factory.h | 43 ++
.../index/pksorted/pk_sorted_index_builder.cpp | 3 +-
src/paimon/core/utils/file_store_path_factory.cpp | 73 ++++
src/paimon/core/utils/file_store_path_factory.h | 5 +
.../core/utils/file_store_path_factory_test.cpp | 127 ++++++
.../lucene/lucene_global_index_test.cpp | 6 +-
src/paimon/global_index/lumina/CMakeLists.txt | 4 +-
.../lumina/lumina_checkpoint_manager.cpp | 55 +++
.../lumina/lumina_checkpoint_manager.h | 45 +++
.../global_index/lumina/lumina_global_index.cpp | 150 +++++--
.../global_index/lumina/lumina_global_index.h | 36 +-
.../lumina/lumina_global_index_test.cpp | 449 ++++++++++++++++++++-
.../tantivy/tantivy_equivalence_test.cpp | 6 +-
.../tantivy/tantivy_filter_limit_test.cpp | 3 +-
.../global_index/tantivy/tantivy_index_test.cpp | 6 +-
.../tantivy/tantivy_java_compat_test.cpp | 6 +-
.../tantivy/tantivy_lucene_coexist_test.cpp | 6 +-
.../tantivy/tantivy_streaming_test.cpp | 6 +-
.../global_index/tantivy/tantivy_writer_test.cpp | 6 +-
test/inte/global_index_test.cpp | 94 ++++-
31 files changed, 1560 insertions(+), 111 deletions(-)
diff --git a/include/paimon/global_index/global_index_write_task.h
b/include/paimon/global_index/global_index_write_task.h
index 33760405..391c02ca 100644
--- a/include/paimon/global_index/global_index_write_task.h
+++ b/include/paimon/global_index/global_index_write_task.h
@@ -21,6 +21,7 @@
#include <map>
#include <memory>
+#include <optional>
#include <string>
#include "paimon/global_index/indexed_split.h"
@@ -45,7 +46,13 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
/// The range must be fully contained within the data
covered
/// by the given `indexed_split`.
/// @param options Index-specific configuration (e.g., false positive
rate for bloom
- /// filters).
+ /// filters).
+ /// @param task_id When checkpoints are enabled, the caller must
provide a non-empty task
+ /// identifier that uniquely identifies an index build
task. Reuse it when
+ /// retrying the same build. If the source data, build
configuration, or
+ /// build source code changes, the caller must use a
new identifier;
+ /// otherwise, the index build may fail. Pass nullopt
when checkpoints are
+ /// disabled. Index types without checkpoint support
ignore this value.
/// @param pool Memory pool for temporary allocations during index
construction.
/// If `nullptr`, the system's default memory pool
will be used.
/// @param file_system Specifies the file system for file operations.
@@ -55,7 +62,8 @@ class PAIMON_EXPORT GlobalIndexWriteTask {
static Result<std::shared_ptr<CommitMessage>> WriteIndex(
const std::string& table_path, const std::string& field_name, const
std::string& index_type,
const std::shared_ptr<IndexedSplit>& indexed_split,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<MemoryPool>& pool,
+ const std::map<std::string, std::string>& options,
+ const std::optional<std::string>& task_id, const
std::shared_ptr<MemoryPool>& pool,
const std::shared_ptr<FileSystem>& file_system = nullptr);
};
diff --git a/include/paimon/global_index/global_indexer.h
b/include/paimon/global_index/global_indexer.h
index 4da6293f..5be789ac 100644
--- a/include/paimon/global_index/global_indexer.h
+++ b/include/paimon/global_index/global_indexer.h
@@ -70,6 +70,11 @@ class PAIMON_EXPORT GlobalIndexer {
::ArrowSchema* arrow_schema, const
std::shared_ptr<GlobalIndexFileReader>& file_reader,
const std::vector<GlobalIndexIOMeta>& files,
const std::shared_ptr<MemoryPool>& pool) const = 0;
+
+ /// Whether this indexer supports checkpointing an index build.
+ virtual bool SupportsCheckpoint() const {
+ return false;
+ }
};
} // namespace paimon
diff --git
a/include/paimon/global_index/io/global_index_checkpoint_file_manager.h
b/include/paimon/global_index/io/global_index_checkpoint_file_manager.h
new file mode 100644
index 00000000..7a07b9ae
--- /dev/null
+++ b/include/paimon/global_index/io/global_index_checkpoint_file_manager.h
@@ -0,0 +1,53 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+class InputStream;
+class OutputStream;
+
+/// Abstract interface for managing checkpoints belonging to one global index
build identity.
+class PAIMON_EXPORT GlobalIndexCheckpointFileManager {
+ public:
+ virtual ~GlobalIndexCheckpointFileManager() = default;
+
+ /// Returns whether checkpoint storage is configured, without accessing
storage.
+ virtual bool SupportsCheckpoint() const = 0;
+
+ /// Creates a new checkpoint file and opens it for writing.
+ virtual Result<std::unique_ptr<OutputStream>>
CreateCheckpointOutputStream() const = 0;
+
+ /// Opens the matching checkpoint with the largest numeric id for reading.
+ virtual Result<std::unique_ptr<InputStream>> OpenCheckpointInputStream()
const = 0;
+
+ /// Returns whether the checkpoint file exists.
+ virtual Result<bool> CheckpointExists() const = 0;
+
+ /// Deletes all matching checkpoint files. Deleting missing checkpoints
succeeds.
+ virtual Status DeleteCheckpoint() const = 0;
+};
+
+} // namespace paimon
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index ffd96c70..6fc99343 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -822,6 +822,7 @@ if(PAIMON_BUILD_TESTS)
core/io/file_index_evaluator_test.cpp
core/io/single_file_writer_test.cpp
core/io/rolling_blob_file_writer_test.cpp
+ core/global_index/global_index_file_manager_test.cpp
core/global_index/global_index_evaluator_impl_test.cpp
core/global_index/indexed_split_test.cpp
core/manifest/file_source_test.cpp
diff --git a/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
b/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
index 80e023f9..6e1ad44a 100644
--- a/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
+++ b/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
@@ -67,7 +67,8 @@ class BitmapGlobalIndexTest : public ::testing::Test {
auto global_index = std::make_shared<BitmapGlobalIndex>(file_index);
auto path_factory = std::make_shared<MockIndexPathFactory>(index_root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<GlobalIndexWriter> global_writer,
@@ -112,7 +113,8 @@ class BitmapGlobalIndexTest : public ::testing::Test {
auto global_index = std::make_shared<BitmapGlobalIndex>(file_index);
auto path_factory = std::make_shared<MockIndexPathFactory>(index_root);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
EXPECT_OK_AND_ASSIGN(
auto global_index_reader,
global_index->CreateReader(CreateArrowSchema(type).get(),
file_reader, {meta}, pool_));
diff --git a/src/paimon/common/global_index/global_indexer_factory_test.cpp
b/src/paimon/common/global_index/global_indexer_factory_test.cpp
index c03fec58..30c58b83 100644
--- a/src/paimon/common/global_index/global_indexer_factory_test.cpp
+++ b/src/paimon/common/global_index/global_indexer_factory_test.cpp
@@ -39,6 +39,7 @@ TEST(GlobalIndexerFactoryTest,
TestLegacyBitmapEnabledForTesting) {
ASSERT_OK_AND_ASSIGN(std::unique_ptr<GlobalIndexer> indexer,
GlobalIndexerFactory::Get("bitmap", options));
ASSERT_TRUE(dynamic_cast<BitmapGlobalIndex*>(indexer.get()));
+ ASSERT_FALSE(indexer->SupportsCheckpoint());
}
TEST(GlobalIndexerFactoryTest, TestNonExist) {
diff --git
a/src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
b/src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
index d19622ad..c198fbf7 100644
---
a/src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
+++
b/src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp
@@ -66,7 +66,8 @@ class RangeBitmapGlobalIndexTest : public ::testing::Test {
auto global_index =
std::make_shared<RangeBitmapGlobalIndex>(file_index);
auto path_factory = std::make_shared<MockIndexPathFactory>(index_root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<GlobalIndexWriter> global_writer,
@@ -109,7 +110,8 @@ class RangeBitmapGlobalIndexTest : public ::testing::Test {
auto global_index =
std::make_shared<RangeBitmapGlobalIndex>(file_index);
auto path_factory = std::make_shared<MockIndexPathFactory>(index_root);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
EXPECT_OK_AND_ASSIGN(
auto global_index_reader,
global_index->CreateReader(CreateArrowSchema(type).get(),
file_reader, {meta}, pool_));
diff --git a/src/paimon/core/global_index/global_index_file_manager.h
b/src/paimon/core/global_index/global_index_file_manager.h
index 3db9965b..8ca5ebf7 100644
--- a/src/paimon/core/global_index/global_index_file_manager.h
+++ b/src/paimon/core/global_index/global_index_file_manager.h
@@ -19,22 +19,37 @@
#pragma once
+#include <algorithm>
+#include <cstdint>
+#include <limits>
#include <memory>
+#include <optional>
#include <string>
+#include <utility>
+#include <vector>
+#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/uuid.h"
+#include "paimon/core/index/index_checkpoint_path_factory.h"
#include "paimon/core/index/index_path_factory.h"
#include "paimon/fs/file_system.h"
+#include "paimon/global_index/io/global_index_checkpoint_file_manager.h"
#include "paimon/global_index/io/global_index_file_reader.h"
#include "paimon/global_index/io/global_index_file_writer.h"
namespace paimon {
/// Helper class for managing global index files.
-class GlobalIndexFileManager : public GlobalIndexFileReader, public
GlobalIndexFileWriter {
+/// Checkpoint storage is optional and is never accessed by construction or
ordinary index I/O.
+class GlobalIndexFileManager : public GlobalIndexFileReader,
+ public GlobalIndexFileWriter,
+ public GlobalIndexCheckpointFileManager {
public:
GlobalIndexFileManager(const std::shared_ptr<FileSystem>& fs,
- const std::shared_ptr<IndexPathFactory>&
path_factory)
- : fs_(fs), path_factory_(path_factory) {}
+ const std::shared_ptr<IndexPathFactory>&
path_factory,
+ std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_path_factory)
+ : fs_(fs),
+ path_factory_(path_factory),
+ checkpoint_path_factory_(std::move(checkpoint_path_factory)) {}
Result<std::unique_ptr<InputStream>> GetInputStream(
const std::string& file_path) const override {
@@ -71,8 +86,110 @@ class GlobalIndexFileManager : public
GlobalIndexFileReader, public GlobalIndexF
return path_factory_->IsExternalPath();
}
+ bool SupportsCheckpoint() const override {
+ return checkpoint_path_factory_ != nullptr;
+ }
+
+ Result<std::unique_ptr<OutputStream>> CreateCheckpointOutputStream() const
override {
+ if (!SupportsCheckpoint()) {
+ return Status::Invalid("global index checkpoint storage is not
configured");
+ }
+ PAIMON_ASSIGN_OR_RAISE(int64_t file_id, NextCheckpointFileId());
+
PAIMON_RETURN_NOT_OK(fs_->Mkdirs(checkpoint_path_factory_->GetDirectoryPath()));
+ return fs_->Create(checkpoint_path_factory_->NewPath(file_id),
/*overwrite=*/false);
+ }
+
+ Result<std::unique_ptr<InputStream>> OpenCheckpointInputStream() const
override {
+ if (!SupportsCheckpoint()) {
+ return Status::Invalid("global index checkpoint storage is not
configured");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::optional<CheckpointFile> checkpoint_file,
+ LatestCheckpointFile());
+ if (!checkpoint_file) {
+ return Status::NotExist("global index checkpoint file does not
exist");
+ }
+ return fs_->Open(checkpoint_file->path);
+ }
+
+ Result<bool> CheckpointExists() const override {
+ if (!SupportsCheckpoint()) {
+ return Status::Invalid("global index checkpoint storage is not
configured");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::optional<CheckpointFile> checkpoint_file,
+ LatestCheckpointFile());
+ return checkpoint_file.has_value();
+ }
+
+ Status DeleteCheckpoint() const override {
+ if (!SupportsCheckpoint()) {
+ return Status::Invalid("global index checkpoint storage is not
configured");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<CheckpointFile> checkpoint_files,
ListCheckpointFiles());
+ Status first_error = Status::OK();
+ for (const CheckpointFile& checkpoint_file : checkpoint_files) {
+ Status status = fs_->Delete(checkpoint_file.path,
/*recursive=*/false);
+ if (!status.ok() && first_error.ok()) {
+ first_error = std::move(status);
+ }
+ }
+ return first_error;
+ }
+
private:
+ struct CheckpointFile {
+ int64_t id;
+ std::string path;
+ };
+
+ Result<int64_t> NextCheckpointFileId() const {
+ if (!last_checkpoint_file_id_) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<CheckpointFile>
checkpoint_file,
+ LatestCheckpointFile());
+ last_checkpoint_file_id_ = checkpoint_file ? checkpoint_file->id :
-1;
+ }
+ if (last_checkpoint_file_id_.value() ==
std::numeric_limits<int64_t>::max()) {
+ return Status::Invalid("checkpoint file id exceeds int64 max");
+ }
+ return ++last_checkpoint_file_id_.value();
+ }
+
+ Result<std::vector<CheckpointFile>> ListCheckpointFiles() const {
+ std::vector<BasicFileStatus> file_statuses;
+ PAIMON_RETURN_NOT_OK(
+ fs_->ListDir(checkpoint_path_factory_->GetDirectoryPath(),
&file_statuses));
+ std::vector<CheckpointFile> checkpoint_files;
+ for (const BasicFileStatus& file_status : file_statuses) {
+ if (file_status.IsDir()) {
+ continue;
+ }
+ std::string file_name = PathUtil::GetName(file_status.GetPath());
+ std::optional<int64_t> id =
checkpoint_path_factory_->GetCheckpointId(file_name);
+ if (!id) {
+ continue;
+ }
+ checkpoint_files.push_back(
+ CheckpointFile{id.value(),
checkpoint_path_factory_->ToPath(file_name)});
+ }
+ return checkpoint_files;
+ }
+
+ Result<std::optional<CheckpointFile>> LatestCheckpointFile() const {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<CheckpointFile> checkpoint_files,
ListCheckpointFiles());
+ if (checkpoint_files.empty()) {
+ return std::optional<CheckpointFile>();
+ }
+ CheckpointFile latest =
+ *std::max_element(checkpoint_files.begin(), checkpoint_files.end(),
+ [](const CheckpointFile& left, const
CheckpointFile& right) {
+ return left.id < right.id;
+ });
+ return std::optional<CheckpointFile>(std::move(latest));
+ }
+
std::shared_ptr<FileSystem> fs_;
std::shared_ptr<IndexPathFactory> path_factory_;
+ std::unique_ptr<IndexCheckpointPathFactory> checkpoint_path_factory_;
+ // Historical ids are loaded only on the first successful file id
allocation scan.
+ mutable std::optional<int64_t> last_checkpoint_file_id_;
};
} // namespace paimon
diff --git a/src/paimon/core/global_index/global_index_file_manager_test.cpp
b/src/paimon/core/global_index/global_index_file_manager_test.cpp
new file mode 100644
index 00000000..ab27ad2e
--- /dev/null
+++ b/src/paimon/core/global_index/global_index_file_manager_test.cpp
@@ -0,0 +1,294 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/global_index/global_index_file_manager.h"
+
+#include <limits>
+#include <set>
+#include <vector>
+
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "gtest/gtest.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/core/utils/file_store_path_factory.h"
+#include "paimon/fs/local/local_file_system.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/utils/range.h"
+
+namespace paimon::test {
+
+class GlobalIndexFileManagerTest : public ::testing::Test {
+ public:
+ class TrackingFileSystem : public LocalFileSystem {
+ public:
+ Status ListDir(const std::string& directory,
+ std::vector<BasicFileStatus>* status_list) const
override {
+ ++list_count_;
+ if (fail_list_) {
+ return Status::IOError("checkpoint list failed");
+ }
+ return LocalFileSystem::ListDir(directory, status_list);
+ }
+ bool fail_list_ = false;
+ mutable int32_t list_count_ = 0;
+ };
+
+ void SetUp() override {
+ dir_ = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir_);
+ 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=*/{},
+ /*global_index_external_path=*/std::nullopt,
+ /*index_file_in_data_file_dir=*/false, GetDefaultPool()));
+ }
+
+ Result<std::shared_ptr<GlobalIndexFileManager>> CreateManager() const {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_path_factory,
+
path_factory_->CreateGlobalIndexCheckpointPathFactory(
+ "lumina", "vector", Range(10, 20),
"task-1"));
+ return std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory_->CreateGlobalIndexFileFactory(),
std::move(checkpoint_path_factory));
+ }
+
+ Result<std::string> CreateCheckpointFile(
+ const std::shared_ptr<GlobalIndexFileManager>& manager) const {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<OutputStream> output,
+ manager->CreateCheckpointOutputStream());
+ PAIMON_ASSIGN_OR_RAISE(std::string uri, output->GetUri());
+ PAIMON_RETURN_NOT_OK(output->Close());
+ return PathUtil::GetName(uri);
+ }
+
+ std::shared_ptr<TrackingFileSystem> fs_ =
std::make_shared<TrackingFileSystem>();
+ std::unique_ptr<UniqueTestDirectory> dir_;
+ std::shared_ptr<FileStorePathFactory> path_factory_;
+};
+
+TEST_F(GlobalIndexFileManagerTest, TestIndexIOWithoutCheckpointAccess) {
+ fs_->fail_list_ = true;
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager,
CreateManager());
+ std::shared_ptr<GlobalIndexFileWriter> writer = manager;
+ std::shared_ptr<GlobalIndexFileReader> reader = manager;
+
ASSERT_TRUE(std::dynamic_pointer_cast<GlobalIndexCheckpointFileManager>(writer));
+
ASSERT_TRUE(std::dynamic_pointer_cast<GlobalIndexCheckpointFileManager>(reader));
+ ASSERT_TRUE(manager->SupportsCheckpoint());
+ auto plain_manager = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory_->CreateGlobalIndexFileFactory(),
/*checkpoint_path_factory=*/nullptr);
+
ASSERT_TRUE(std::dynamic_pointer_cast<GlobalIndexCheckpointFileManager>(plain_manager));
+ ASSERT_FALSE(plain_manager->SupportsCheckpoint());
+
+ ASSERT_OK_AND_ASSIGN(std::string name, writer->NewFileName("lumina"));
+ ASSERT_TRUE(StringUtils::EndsWith(name, ".index"));
+ ASSERT_EQ(writer->ToPath(name), PathUtil::JoinPath(dir_->Str(), "index/" +
name));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> output,
writer->NewOutputStream(name));
+ ASSERT_OK_AND_ASSIGN(int64_t written, output->Write("index", 5));
+ ASSERT_EQ(written, 5);
+ ASSERT_OK(output->Close());
+ ASSERT_OK_AND_ASSIGN(int64_t size, writer->GetFileSize(name));
+ ASSERT_EQ(size, 5);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input,
+ reader->GetInputStream(writer->ToPath(name)));
+ char buffer[5];
+ ASSERT_OK_AND_ASSIGN(int64_t read, input->Read(buffer, sizeof(buffer)));
+ ASSERT_EQ(read, 5);
+ ASSERT_EQ(std::string(buffer, sizeof(buffer)), "index");
+ ASSERT_OK(input->Close());
+ ASSERT_EQ(fs_->list_count_, 0);
+ ASSERT_OK_AND_ASSIGN(bool exists,
+ fs_->Exists(PathUtil::JoinPath(dir_->Str(),
"index/checkpoint")));
+ ASSERT_FALSE(exists);
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestCheckpointNotConfigured) {
+ fs_->fail_list_ = true;
+ GlobalIndexFileManager manager(fs_,
path_factory_->CreateGlobalIndexFileFactory(),
+ /*checkpoint_path_factory=*/nullptr);
+ ASSERT_FALSE(manager.SupportsCheckpoint());
+ ASSERT_NOK_WITH_MSG(manager.CreateCheckpointOutputStream(),
+ "checkpoint storage is not configured");
+ ASSERT_NOK_WITH_MSG(manager.OpenCheckpointInputStream(),
+ "checkpoint storage is not configured");
+ ASSERT_NOK_WITH_MSG(manager.CheckpointExists(), "checkpoint storage is not
configured");
+ ASSERT_NOK_WITH_MSG(manager.DeleteCheckpoint(), "checkpoint storage is not
configured");
+ ASSERT_EQ(fs_->list_count_, 0);
+ ASSERT_OK_AND_ASSIGN(bool exists,
+ fs_->Exists(PathUtil::JoinPath(dir_->Str(),
"index/checkpoint")));
+ ASSERT_FALSE(exists);
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestCheckpointIOAndRestart) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager,
CreateManager());
+ ASSERT_EQ(fs_->list_count_, 0);
+ std::string directory = PathUtil::JoinPath(dir_->Str(),
"index/checkpoint");
+ std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+ std::string previous_path = PathUtil::JoinPath(directory, prefix +
"9.index.ckpt");
+ // The latest sequence must be discovered on first file name allocation,
not at construction.
+ ASSERT_OK(fs_->WriteFile(previous_path, "old", /*overwrite=*/false));
+ ASSERT_OK_AND_ASSIGN(std::string name, CreateCheckpointFile(manager));
+ ASSERT_TRUE(StringUtils::StartsWith(name, prefix));
+ ASSERT_TRUE(StringUtils::EndsWith(name, "-10.index.ckpt"));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> output,
+ manager->CreateCheckpointOutputStream());
+ ASSERT_OK_AND_ASSIGN(int64_t written, output->Write("latest", 6));
+ ASSERT_EQ(written, 6);
+ ASSERT_OK(output->Close());
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> restarted,
CreateManager());
+ ASSERT_OK_AND_ASSIGN(bool exists, restarted->CheckpointExists());
+ ASSERT_TRUE(exists);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input,
+ restarted->OpenCheckpointInputStream());
+ char buffer[6];
+ ASSERT_OK_AND_ASSIGN(int64_t read, input->Read(buffer, sizeof(buffer)));
+ ASSERT_EQ(read, 6);
+ ASSERT_EQ(std::string(buffer, sizeof(buffer)), "latest");
+ ASSERT_OK(input->Close());
+ ASSERT_OK_AND_ASSIGN(name, CreateCheckpointFile(restarted));
+ ASSERT_TRUE(StringUtils::EndsWith(name, "-12.index.ckpt"));
+
+ std::string other_task_path =
+ PathUtil::JoinPath(directory,
"lumina-global-index-vector-10-20-task-2-99.index.ckpt");
+ std::string index_path = manager->ToPath("retained.index");
+ ASSERT_OK(fs_->WriteFile(other_task_path, "other", /*overwrite=*/false));
+ ASSERT_OK(fs_->WriteFile(index_path, "index", /*overwrite=*/false));
+ ASSERT_OK(restarted->DeleteCheckpoint());
+ ASSERT_OK_AND_ASSIGN(exists, restarted->CheckpointExists());
+ ASSERT_FALSE(exists);
+ ASSERT_NOK_WITH_MSG(restarted->OpenCheckpointInputStream(), "checkpoint
file does not exist");
+ ASSERT_OK(restarted->DeleteCheckpoint());
+ ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(previous_path));
+ ASSERT_FALSE(exists);
+ ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(other_task_path));
+ ASSERT_TRUE(exists);
+ ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(index_path));
+ ASSERT_TRUE(exists);
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestInitializationFailureCanRetry) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager,
CreateManager());
+ fs_->fail_list_ = true;
+ ASSERT_TRUE(manager->SupportsCheckpoint());
+ ASSERT_EQ(fs_->list_count_, 0);
+ ASSERT_NOK_WITH_MSG(CreateCheckpointFile(manager), "checkpoint list
failed");
+ ASSERT_NOK_WITH_MSG(manager->CreateCheckpointOutputStream(), "checkpoint
list failed");
+ ASSERT_NOK_WITH_MSG(manager->OpenCheckpointInputStream(), "checkpoint list
failed");
+ ASSERT_NOK_WITH_MSG(manager->CheckpointExists(), "checkpoint list failed");
+ ASSERT_NOK_WITH_MSG(manager->DeleteCheckpoint(), "checkpoint list failed");
+ fs_->fail_list_ = false;
+ ASSERT_OK_AND_ASSIGN(std::string first, CreateCheckpointFile(manager));
+ ASSERT_TRUE(StringUtils::EndsWith(first, "-0.index.ckpt"));
+ ASSERT_OK_AND_ASSIGN(std::string second, CreateCheckpointFile(manager));
+ ASSERT_TRUE(StringUtils::EndsWith(second, "-1.index.ckpt"));
+ ASSERT_EQ(fs_->list_count_, 6);
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestLatestCheckpointUsesNumericId) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager,
CreateManager());
+ std::string directory = PathUtil::JoinPath(dir_->Str(),
"index/checkpoint");
+ std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+ ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix +
"9.index.ckpt"), "older",
+ /*overwrite=*/false));
+ ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix +
"10.index.ckpt"), "newer",
+ /*overwrite=*/false));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input,
manager->OpenCheckpointInputStream());
+ char buffer[5];
+ ASSERT_OK_AND_ASSIGN(int64_t read, input->Read(buffer, sizeof(buffer)));
+ ASSERT_EQ(read, 5);
+ ASSERT_EQ(std::string(buffer, sizeof(buffer)), "newer");
+ ASSERT_OK(input->Close());
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestReadsDoNotInitializeFileId) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager,
CreateManager());
+ ASSERT_OK_AND_ASSIGN(bool exists, manager->CheckpointExists());
+ ASSERT_FALSE(exists);
+ std::string directory = PathUtil::JoinPath(dir_->Str(),
"index/checkpoint");
+ std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+ ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix +
"9.index.ckpt"), "old",
+ /*overwrite=*/false));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input,
manager->OpenCheckpointInputStream());
+ ASSERT_OK(input->Close());
+ ASSERT_EQ(fs_->list_count_, 2);
+
+ ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(directory, prefix +
"19.index.ckpt"), "latest",
+ /*overwrite=*/false));
+ ASSERT_OK_AND_ASSIGN(std::string first, CreateCheckpointFile(manager));
+ ASSERT_TRUE(StringUtils::EndsWith(first, "-20.index.ckpt"));
+ ASSERT_EQ(fs_->list_count_, 3);
+ ASSERT_OK_AND_ASSIGN(std::string second, CreateCheckpointFile(manager));
+ ASSERT_TRUE(StringUtils::EndsWith(second, "-21.index.ckpt"));
+ ASSERT_EQ(fs_->list_count_, 3);
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestFileIdOverflow) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager,
CreateManager());
+ std::string directory = PathUtil::JoinPath(dir_->Str(),
"index/checkpoint");
+ std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+ int64_t max_id = std::numeric_limits<int64_t>::max();
+ ASSERT_OK(fs_->WriteFile(
+ PathUtil::JoinPath(directory, fmt::format("{}{}.index.ckpt", prefix,
max_id - 1)), "old",
+ /*overwrite=*/false));
+ ASSERT_OK_AND_ASSIGN(std::string name, CreateCheckpointFile(manager));
+ ASSERT_TRUE(StringUtils::EndsWith(name, fmt::format("-{}.index.ckpt",
max_id)));
+ ASSERT_NOK_WITH_MSG(CreateCheckpointFile(manager), "checkpoint file id
exceeds int64 max");
+ ASSERT_NOK_WITH_MSG(manager->CreateCheckpointOutputStream(),
+ "checkpoint file id exceeds int64 max");
+ ASSERT_EQ(fs_->list_count_, 1);
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> restarted,
CreateManager());
+ ASSERT_EQ(fs_->list_count_, 1);
+ ASSERT_NOK_WITH_MSG(CreateCheckpointFile(restarted), "checkpoint file id
exceeds int64 max");
+ ASSERT_NOK_WITH_MSG(CreateCheckpointFile(restarted), "checkpoint file id
exceeds int64 max");
+ ASSERT_EQ(fs_->list_count_, 2);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input,
+ restarted->OpenCheckpointInputStream());
+ ASSERT_OK(input->Close());
+ ASSERT_OK(restarted->DeleteCheckpoint());
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestSerialAllocations) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager,
CreateManager());
+ std::vector<std::string> names(8);
+ for (size_t i = 0; i < names.size(); ++i) {
+ ASSERT_OK_AND_ASSIGN(names[i], CreateCheckpointFile(manager));
+ ASSERT_TRUE(StringUtils::EndsWith(names[i],
fmt::format("-{}.index.ckpt", i)));
+ }
+ ASSERT_EQ(fs_->list_count_, 1);
+ ASSERT_EQ(std::set<std::string>(names.begin(), names.end()).size(),
names.size());
+ for (const std::string& name : names) {
+ ASSERT_FALSE(name.empty());
+ }
+}
+
+TEST_F(GlobalIndexFileManagerTest, TestDeleteDoesNotResetFileId) {
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexFileManager> manager,
CreateManager());
+ ASSERT_OK_AND_ASSIGN(std::string first, CreateCheckpointFile(manager));
+ ASSERT_TRUE(StringUtils::EndsWith(first, "-0.index.ckpt"));
+ ASSERT_OK(manager->DeleteCheckpoint());
+ ASSERT_OK_AND_ASSIGN(std::string second, CreateCheckpointFile(manager));
+ ASSERT_TRUE(StringUtils::EndsWith(second, "-1.index.ckpt"));
+ ASSERT_EQ(fs_->list_count_, 2);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/global_index/global_index_scan_impl.cpp
b/src/paimon/core/global_index/global_index_scan_impl.cpp
index e4b3e7b7..c38c5037 100644
--- a/src/paimon/core/global_index/global_index_scan_impl.cpp
+++ b/src/paimon/core/global_index/global_index_scan_impl.cpp
@@ -43,8 +43,8 @@ GlobalIndexScanImpl::GlobalIndexScanImpl(const
std::shared_ptr<TableSchema>& tab
: pool_(pool),
table_schema_(table_schema),
options_(options),
- index_file_manager_(
- std::make_shared<GlobalIndexFileManager>(options.GetFileSystem(),
path_factory)),
+ index_file_manager_(std::make_shared<GlobalIndexFileManager>(
+ options.GetFileSystem(), path_factory,
/*checkpoint_path_factory=*/nullptr)),
index_metas_(std::move(index_metas)),
executor_(executor) {}
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 0288d1b9..82c86143 100644
--- a/src/paimon/core/global_index/global_index_write_task.cpp
+++ b/src/paimon/core/global_index/global_index_write_task.cpp
@@ -39,6 +39,7 @@
#include "paimon/core/schema/table_schema.h"
#include "paimon/core/table/sink/commit_message_impl.h"
#include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/utils/branch_manager.h"
#include "paimon/core/utils/file_store_path_factory.h"
#include "paimon/global_index/global_indexer.h"
#include "paimon/global_index/global_indexer_factory.h"
@@ -57,7 +58,7 @@ Result<std::unique_ptr<GlobalIndexer>>
CreateGlobalIndexer(const std::string& in
return indexer;
}
-Result<std::shared_ptr<GlobalIndexFileManager>> CreateGlobalIndexFileManager(
+Result<std::shared_ptr<FileStorePathFactory>> CreateFileStorePathFactory(
const std::string& table_path, const std::shared_ptr<TableSchema>&
table_schema,
const CoreOptions& core_options, const std::shared_ptr<MemoryPool>& pool) {
auto all_arrow_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
@@ -65,18 +66,11 @@ Result<std::shared_ptr<GlobalIndexFileManager>>
CreateGlobalIndexFileManager(
core_options.CreateExternalPaths());
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string>
global_index_external_path,
core_options.CreateGlobalIndexExternalPath());
- PAIMON_ASSIGN_OR_RAISE(
- std::shared_ptr<FileStorePathFactory> path_factory,
- FileStorePathFactory::Create(
- table_path, all_arrow_schema, table_schema->PartitionKeys(),
- core_options.GetPartitionDefaultName(),
core_options.GetFileFormat()->Identifier(),
- core_options.DataFilePrefix(),
core_options.LegacyPartitionNameEnabled(),
- external_paths, global_index_external_path,
core_options.IndexFileInDataFileDir(),
- pool));
- std::shared_ptr<IndexPathFactory> index_path_factory =
- path_factory->CreateGlobalIndexFileFactory();
- return
std::make_shared<GlobalIndexFileManager>(core_options.GetFileSystem(),
- index_path_factory);
+ return FileStorePathFactory::Create(
+ table_path, all_arrow_schema, table_schema->PartitionKeys(),
+ core_options.GetPartitionDefaultName(),
core_options.GetFileFormat()->Identifier(),
+ core_options.DataFilePrefix(),
core_options.LegacyPartitionNameEnabled(), external_paths,
+ global_index_external_path, core_options.IndexFileInDataFileDir(),
pool);
}
Result<std::shared_ptr<GlobalIndexWriter>> CreateGlobalIndexWriter(
@@ -344,7 +338,7 @@ Result<std::shared_ptr<CommitMessage>> ToCommitMessage(
Result<std::shared_ptr<CommitMessage>> GlobalIndexWriteTask::WriteIndex(
const std::string& table_path, const std::string& field_name, const
std::string& index_type,
const std::shared_ptr<IndexedSplit>& indexed_split,
- const std::map<std::string, std::string>& options,
+ const std::map<std::string, std::string>& options, const
std::optional<std::string>& task_id,
const std::shared_ptr<MemoryPool>& memory_pool,
const std::shared_ptr<FileSystem>& file_system) {
auto data_split =
std::dynamic_pointer_cast<DataSplitImpl>(indexed_split->GetDataSplit());
@@ -387,10 +381,19 @@ Result<std::shared_ptr<CommitMessage>>
GlobalIndexWriteTask::WriteIndex(
std::vector<std::string> writer_field_names =
BuildWriterFieldNames(field_name, extra_fields);
std::vector<std::string> read_field_names =
BuildReadFieldNames(field_name, extra_fields);
- // create index file manager
+ // Checkpoint capability is optional; only plugins enabling checkpoints
will use it.
PAIMON_ASSIGN_OR_RAISE(
- std::shared_ptr<GlobalIndexFileManager> index_file_manager,
- CreateGlobalIndexFileManager(table_path, table_schema, core_options,
pool));
+ std::shared_ptr<FileStorePathFactory> path_factory,
+ CreateFileStorePathFactory(table_path, table_schema, core_options,
pool));
+ std::unique_ptr<IndexCheckpointPathFactory> checkpoint_path_factory;
+ if (task_id && !task_id->empty() && indexer->SupportsCheckpoint()) {
+ PAIMON_ASSIGN_OR_RAISE(checkpoint_path_factory,
+
path_factory->CreateGlobalIndexCheckpointPathFactory(
+ index_type, field_name, range,
task_id.value()));
+ }
+ auto index_file_manager = std::make_shared<GlobalIndexFileManager>(
+ core_options.GetFileSystem(),
path_factory->CreateGlobalIndexFileFactory(),
+ std::move(checkpoint_path_factory));
// create batch reader
PAIMON_ASSIGN_OR_RAISE(
diff --git a/src/paimon/core/index/index_checkpoint_path_factory.h
b/src/paimon/core/index/index_checkpoint_path_factory.h
new file mode 100644
index 00000000..0c1b13e5
--- /dev/null
+++ b/src/paimon/core/index/index_checkpoint_path_factory.h
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <optional>
+#include <string>
+
+namespace paimon {
+
+/// Path factory for global index checkpoints scoped to one task.
+class IndexCheckpointPathFactory {
+ public:
+ virtual ~IndexCheckpointPathFactory() = default;
+
+ /// Creates the path for the specified checkpoint id, without storage I/O.
+ virtual std::string NewPath(int64_t checkpoint_id) const = 0;
+ virtual std::string ToPath(const std::string& file_name) const = 0;
+
+ /// Returns the directory containing the checkpoint files.
+ virtual const std::string& GetDirectoryPath() const = 0;
+
+ /// 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;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/index/pksorted/pk_sorted_index_builder.cpp
b/src/paimon/core/index/pksorted/pk_sorted_index_builder.cpp
index 765c1ea3..4d36aa1f 100644
--- a/src/paimon/core/index/pksorted/pk_sorted_index_builder.cpp
+++ b/src/paimon/core/index/pksorted/pk_sorted_index_builder.cpp
@@ -161,7 +161,8 @@ Result<std::shared_ptr<IndexFileMeta>>
PkSortedIndexBuilder::Build(
auto sorted_reader = std::make_unique<SortMergeReaderWithMinHeap>(
std::move(readers), comparator, sequence_comparator,
/*merge_function_wrapper=*/nullptr);
- auto file_manager = std::make_shared<GlobalIndexFileManager>(fs_,
index_path_factory_);
+ auto file_manager = std::make_shared<GlobalIndexFileManager>(
+ fs_, index_path_factory_, /*checkpoint_path_factory=*/nullptr);
auto tracking_writer =
std::make_shared<TrackingGlobalIndexFileWriter>(file_manager);
Result<std::shared_ptr<IndexFileMeta>> result =
PkSortedIndexFile::BuildFromSortedReader(
field_, definition_.IndexType(), definition_.Options(), data_level,
source_metas,
diff --git a/src/paimon/core/utils/file_store_path_factory.cpp
b/src/paimon/core/utils/file_store_path_factory.cpp
index ef4aae42..ac15ad63 100644
--- a/src/paimon/core/utils/file_store_path_factory.cpp
+++ b/src/paimon/core/utils/file_store_path_factory.cpp
@@ -20,8 +20,11 @@
#include <cassert>
+#include "fmt/format.h"
#include "paimon/common/fs/external_path_provider.h"
+#include "paimon/common/utils/string_utils.h"
#include "paimon/common/utils/uuid.h"
+#include "paimon/core/index/index_checkpoint_path_factory.h"
#include "paimon/core/index/index_file_meta.h"
#include "paimon/core/index/index_in_data_file_dir_path_factory.h"
#include "paimon/core/io/data_file_path_factory.h"
@@ -31,6 +34,7 @@
#include "paimon/macros.h"
#include "paimon/memory/memory_segment.h"
#include "paimon/status.h"
+#include "paimon/utils/range.h"
namespace arrow {
class Schema;
@@ -39,6 +43,31 @@ class Schema;
namespace paimon {
class MemoryPool;
+namespace {
+
+constexpr char kIndexCheckpointFileSuffix[] = ".index.ckpt";
+
+std::optional<int64_t> ParseCheckpointId(const std::string& file_name,
+ const std::string& file_name_prefix) {
+ if (!StringUtils::StartsWith(file_name, file_name_prefix) ||
+ !StringUtils::EndsWith(file_name, kIndexCheckpointFileSuffix)) {
+ return std::nullopt;
+ }
+ size_t suffix_pos =
+ file_name.size() -
std::char_traits<char>::length(kIndexCheckpointFileSuffix);
+ if (suffix_pos <= file_name_prefix.size()) {
+ return std::nullopt;
+ }
+ std::optional<int64_t> id = StringUtils::StringToValue<int64_t>(
+ file_name.substr(file_name_prefix.size(), suffix_pos -
file_name_prefix.size()));
+ if (!id || id.value() < 0) {
+ return std::nullopt;
+ }
+ return id;
+}
+
+} // namespace
+
FileStorePathFactory::FileStorePathFactory(
const std::string& root, const std::string& format_identifier,
const std::string& data_file_prefix, const std::string& uuid,
@@ -183,6 +212,50 @@ std::unique_ptr<IndexPathFactory>
FileStorePathFactory::CreateGlobalIndexFileFac
return std::make_unique<IndexPathFactoryImpl>(shared_from_this());
}
+Result<std::unique_ptr<IndexCheckpointPathFactory>>
+FileStorePathFactory::CreateGlobalIndexCheckpointPathFactory(const
std::string& index_type,
+ const
std::string& field_name,
+ const Range&
range,
+ const
std::string& task_id) {
+ class IndexCheckpointPathFactoryImpl : public IndexCheckpointPathFactory {
+ public:
+ IndexCheckpointPathFactoryImpl(const std::string& directory,
+ const std::string& file_name_prefix)
+ : directory_(directory), file_name_prefix_(file_name_prefix) {}
+
+ std::string NewPath(int64_t checkpoint_id) const override {
+ assert(checkpoint_id >= 0);
+ std::string file_name =
+ fmt::format("{}{}{}", file_name_prefix_, checkpoint_id,
kIndexCheckpointFileSuffix);
+ return ToPath(file_name);
+ }
+
+ std::string ToPath(const std::string& file_name) const override {
+ return PathUtil::JoinPath(directory_, file_name);
+ }
+
+ const std::string& GetDirectoryPath() const override {
+ return directory_;
+ }
+
+ std::optional<int64_t> GetCheckpointId(const std::string& file_name)
const override {
+ return ParseCheckpointId(file_name, file_name_prefix_);
+ }
+
+ private:
+ std::string directory_;
+ std::string file_name_prefix_;
+ };
+ PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint index
type", index_type));
+ PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint
field", field_name));
+ PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint task
id", task_id));
+ std::string directory = PathUtil::JoinPath(IndexPath(root_), "checkpoint");
+ std::string file_name_prefix = fmt::format("{}-global-index-{}-{}-{}-{}-",
index_type,
+ field_name, range.from,
range.to, task_id);
+ return std::unique_ptr<IndexCheckpointPathFactory>(
+ std::make_unique<IndexCheckpointPathFactoryImpl>(directory,
file_name_prefix));
+}
+
Result<std::shared_ptr<DataFilePathFactory>>
FileStorePathFactory::CreateDataFilePathFactory(
const BinaryRow& partition, int32_t bucket) const {
auto data_file_path_factory = std::make_shared<DataFilePathFactory>();
diff --git a/src/paimon/core/utils/file_store_path_factory.h
b/src/paimon/core/utils/file_store_path_factory.h
index 1890f2bc..030b6b78 100644
--- a/src/paimon/core/utils/file_store_path_factory.h
+++ b/src/paimon/core/utils/file_store_path_factory.h
@@ -33,6 +33,7 @@
#include "paimon/common/data/binary_row.h"
#include "paimon/common/utils/binary_row_partition_computer.h"
#include "paimon/common/utils/path_util.h"
+#include "paimon/core/index/index_checkpoint_path_factory.h"
#include "paimon/core/index/index_path_factory.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/result.h"
@@ -47,6 +48,7 @@ class DataFilePathFactory;
class ExternalPathProvider;
class PathFactory;
class MemoryPool;
+struct Range;
class FileStorePathFactory : public
std::enable_shared_from_this<FileStorePathFactory> {
public:
@@ -76,6 +78,9 @@ class FileStorePathFactory : public
std::enable_shared_from_this<FileStorePathFa
Result<std::unique_ptr<IndexPathFactory>> CreateIndexFileFactory(const
BinaryRow& partition,
int32_t
bucket);
std::unique_ptr<IndexPathFactory> CreateGlobalIndexFileFactory();
+ Result<std::unique_ptr<IndexCheckpointPathFactory>>
CreateGlobalIndexCheckpointPathFactory(
+ const std::string& index_type, const std::string& field_name, const
Range& range,
+ const std::string& task_id);
Result<std::shared_ptr<DataFilePathFactory>> CreateDataFilePathFactory(
const BinaryRow& partition, int32_t bucket) const;
Result<BinaryRow> ToBinaryRow(const std::map<std::string, std::string>&
partition) const;
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 c78db99d..bfb2b962 100644
--- a/src/paimon/core/utils/file_store_path_factory_test.cpp
+++ b/src/paimon/core/utils/file_store_path_factory_test.cpp
@@ -19,6 +19,7 @@
#include "paimon/core/utils/file_store_path_factory.h"
#include <atomic>
+#include <limits>
#include <mutex>
#include <optional>
#include <thread>
@@ -28,14 +29,20 @@
#include "gtest/gtest.h"
#include "paimon/common/data/binary_row_writer.h"
#include "paimon/common/data/data_define.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/common/utils/string_utils.h"
#include "paimon/core/core_options.h"
+#include "paimon/core/global_index/global_index_file_manager.h"
+#include "paimon/core/index/index_checkpoint_path_factory.h"
#include "paimon/core/io/data_file_path_factory.h"
#include "paimon/defs.h"
#include "paimon/format/file_format.h"
+#include "paimon/fs/file_system.h"
#include "paimon/memory/memory_pool.h"
#include "paimon/status.h"
#include "paimon/testing/utils/binary_row_generator.h"
#include "paimon/testing/utils/testharness.h"
+#include "paimon/utils/range.h"
namespace paimon::test {
@@ -582,4 +589,124 @@ TEST_F(FileStorePathFactoryTest,
TestCreateIndexFileFactory) {
ASSERT_EQ(index_path_factory->ToPath(index_file_meta),
"/tmp/external-path/bitmap.index");
}
}
+
+TEST_F(FileStorePathFactoryTest, TestCreateGlobalIndexCheckpointPathFactory) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ std::shared_ptr<FileStorePathFactory> file_store_path_factory =
CreateFactory(dir->Str());
+ std::shared_ptr<FileSystem> file_system = dir->GetFileSystem();
+
+ std::string prefix = "lumina-global-index-vector-10-20-task-1-";
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_path_factory,
+
file_store_path_factory->CreateGlobalIndexCheckpointPathFactory(
+ "lumina", "vector", Range(10, 20), "task-1"));
+ std::string checkpoint_dir = PathUtil::JoinPath(dir->Str(),
"index/checkpoint");
+ ASSERT_EQ(checkpoint_path_factory->GetDirectoryPath(), checkpoint_dir);
+ std::string first_path = checkpoint_path_factory->NewPath(0);
+ ASSERT_EQ(first_path, PathUtil::JoinPath(checkpoint_dir, prefix +
"0.index.ckpt"));
+ std::string second_path = checkpoint_path_factory->NewPath(1);
+ ASSERT_EQ(second_path, PathUtil::JoinPath(checkpoint_dir, prefix +
"1.index.ckpt"));
+ ASSERT_EQ(checkpoint_path_factory->ToPath("checkpoint"),
+ PathUtil::JoinPath(checkpoint_dir, "checkpoint"));
+ ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(prefix +
"9.index.ckpt"), 9);
+ ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(
+ "lumina-global-index-other-10-20-task-1-9.index.ckpt"),
+ std::nullopt);
+ ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(prefix +
"invalid-10.index.ckpt"),
+ std::nullopt);
+ ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(prefix +
"-1.index.ckpt"), std::nullopt);
+ ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(
+ prefix +
"00000000-0000-0000-0000-000000000000-9.index.ckpt"),
+ std::nullopt);
+ ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(
+ "lumina-global-index-7-vector-10_20-0000000000000000-"
+ "00000000-0000-0000-0000-000000000000-9.index.ckpt"),
+ std::nullopt);
+ ASSERT_EQ(checkpoint_path_factory->GetCheckpointId(
+ "lumina-global-index-field=vector-range=10_20-task=task-1-"
+ "00000000-0000-0000-0000-000000000000-9.index.ckpt"),
+ std::nullopt);
+
+ ASSERT_OK_AND_ASSIGN(bool exists, file_system->Exists(checkpoint_dir));
+ ASSERT_FALSE(exists);
+}
+
+TEST_F(FileStorePathFactoryTest, TestCheckpointFileIdPath) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ auto factory = CreateFactory(dir->Str());
+ for (int64_t file_id : {int64_t{9}, std::numeric_limits<int64_t>::max() -
1,
+ std::numeric_limits<int64_t>::max()}) {
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_factory,
+ factory->CreateGlobalIndexCheckpointPathFactory(
+ "lumina", "vector", Range(10, 20), "task-1"));
+ std::string path = checkpoint_factory->NewPath(file_id);
+
ASSERT_EQ(checkpoint_factory->GetCheckpointId(PathUtil::GetName(path)),
file_id);
+ ASSERT_OK_AND_ASSIGN(bool exists,
+
dir->GetFileSystem()->Exists(checkpoint_factory->GetDirectoryPath()));
+ ASSERT_FALSE(exists);
+ }
+}
+
+TEST_F(FileStorePathFactoryTest, TestCheckpointTaskIsolation) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ std::shared_ptr<FileSystem> fs = dir->GetFileSystem();
+ std::shared_ptr<FileStorePathFactory> factory = CreateFactory(dir->Str());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexCheckpointPathFactory>
own_factory,
+
factory->CreateGlobalIndexCheckpointPathFactory("lumina", "vector",
+
Range(10, 20), "task-1"));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexCheckpointPathFactory>
other_factory,
+
factory->CreateGlobalIndexCheckpointPathFactory("lumina", "vector",
+
Range(10, 20), "task-2"));
+ ASSERT_EQ(own_factory->GetDirectoryPath(),
other_factory->GetDirectoryPath());
+ ASSERT_OK(fs->Mkdirs(own_factory->GetDirectoryPath()));
+ std::string own_path = own_factory->NewPath(0);
+ ASSERT_OK(fs->WriteFile(own_path, "checkpoint", /*overwrite=*/false));
+ std::string other_path = other_factory->NewPath(100);
+ ASSERT_OK(fs->WriteFile(other_path, "foreign", /*overwrite=*/false));
+ ASSERT_EQ(own_factory->GetCheckpointId(PathUtil::GetName(own_path)), 0);
+ ASSERT_EQ(own_factory->GetCheckpointId(PathUtil::GetName(other_path)),
std::nullopt);
+
+ GlobalIndexFileManager manager(fs, factory->CreateGlobalIndexFileFactory(),
+ std::move(own_factory));
+ ASSERT_OK_AND_ASSIGN(bool exists, manager.CheckpointExists());
+ ASSERT_TRUE(exists);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> input,
manager.OpenCheckpointInputStream());
+ char buffer[10];
+ ASSERT_OK_AND_ASSIGN(int64_t read, input->Read(buffer, sizeof(buffer)));
+ ASSERT_EQ(read, 10);
+ ASSERT_EQ(std::string(buffer, sizeof(buffer)), "checkpoint");
+ ASSERT_OK(input->Close());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> output,
+ manager.CreateCheckpointOutputStream());
+ ASSERT_OK_AND_ASSIGN(std::string uri, output->GetUri());
+ ASSERT_TRUE(StringUtils::EndsWith(uri, "-1.index.ckpt"));
+ ASSERT_OK(output->Close());
+ ASSERT_OK(manager.DeleteCheckpoint());
+ ASSERT_OK_AND_ASSIGN(exists, manager.CheckpointExists());
+ ASSERT_FALSE(exists);
+ ASSERT_NOK_WITH_MSG(manager.OpenCheckpointInputStream(), "checkpoint file
does not exist");
+ ASSERT_OK(manager.DeleteCheckpoint());
+ ASSERT_OK_AND_ASSIGN(exists, fs->Exists(other_path));
+ ASSERT_TRUE(exists);
+}
+
+TEST_F(FileStorePathFactoryTest, TestCheckpointRejectsInvalidPathComponents) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ 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(factory->CreateGlobalIndexCheckpointPathFactory(invalid,
"vector", Range(10, 20),
+ "task"));
+ ASSERT_NOK(factory->CreateGlobalIndexCheckpointPathFactory("lumina",
invalid, Range(10, 20),
+ "task"));
+ ASSERT_NOK(factory->CreateGlobalIndexCheckpointPathFactory("lumina",
"vector",
+ Range(10,
20), invalid));
+ }
+ ASSERT_OK_AND_ASSIGN(bool exists, dir->GetFileSystem()->Exists(
+ PathUtil::JoinPath(dir->Str(),
"index/checkpoint")));
+ ASSERT_FALSE(exists);
+}
} // namespace paimon::test
diff --git a/src/paimon/global_index/lucene/lucene_global_index_test.cpp
b/src/paimon/global_index/lucene/lucene_global_index_test.cpp
index 1e30b366..a63e96e6 100644
--- a/src/paimon/global_index/lucene/lucene_global_index_test.cpp
+++ b/src/paimon/global_index/lucene/lucene_global_index_test.cpp
@@ -75,7 +75,8 @@ class LuceneGlobalIndexTest : public ::testing::Test,
const std::string& tmp_dir)
const {
auto global_index = std::make_shared<LuceneGlobalIndex>(options);
auto path_factory = std::make_shared<FakeIndexPathFactory>(index_root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexWriter>
global_writer,
global_index->CreateWriter("f0",
CreateArrowSchema(data_type).get(),
@@ -114,7 +115,8 @@ class LuceneGlobalIndexTest : public ::testing::Test,
const std::map<std::string, std::string>& options, const
GlobalIndexIOMeta& meta) const {
auto global_index = std::make_shared<LuceneGlobalIndex>(options);
auto path_factory = std::make_shared<FakeIndexPathFactory>(index_root);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
return global_index->CreateReader(CreateArrowSchema(data_type).get(),
file_reader, {meta},
pool_);
}
diff --git a/src/paimon/global_index/lumina/CMakeLists.txt
b/src/paimon/global_index/lumina/CMakeLists.txt
index b0496df6..239c7702 100644
--- a/src/paimon/global_index/lumina/CMakeLists.txt
+++ b/src/paimon/global_index/lumina/CMakeLists.txt
@@ -15,8 +15,8 @@
# limitations under the License.
if(PAIMON_ENABLE_LUMINA)
- set(PAIMON_LUMINA_INDEX lumina_global_index.cpp
lumina_global_index_factory.cpp)
-
+ set(PAIMON_LUMINA_INDEX lumina_checkpoint_manager.cpp
lumina_global_index.cpp
+ lumina_global_index_factory.cpp)
add_paimon_lib(paimon_lumina_index
SOURCES
${PAIMON_LUMINA_INDEX}
diff --git a/src/paimon/global_index/lumina/lumina_checkpoint_manager.cpp
b/src/paimon/global_index/lumina/lumina_checkpoint_manager.cpp
new file mode 100644
index 00000000..93187eda
--- /dev/null
+++ b/src/paimon/global_index/lumina/lumina_checkpoint_manager.cpp
@@ -0,0 +1,55 @@
+/*
+ * 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.
+ */
+
+#include "paimon/global_index/lumina/lumina_checkpoint_manager.h"
+
+#include <utility>
+
+#include "paimon/global_index/lumina/lumina_file_reader.h"
+#include "paimon/global_index/lumina/lumina_file_writer.h"
+#include "paimon/global_index/lumina/lumina_utils.h"
+
+namespace paimon::lumina {
+
+std::unique_ptr<::lumina::io::FileWriter>
LuminaCheckpointManager::CreateCkptFileWriter() {
+ Result<std::unique_ptr<OutputStream>> output =
file_manager_->CreateCheckpointOutputStream();
+ if (!output.ok()) {
+ return nullptr;
+ }
+ std::shared_ptr<OutputStream> shared_output = std::move(output).value();
+ return std::make_unique<LuminaFileWriter>(shared_output);
+}
+
+::lumina::core::Result<bool> LuminaCheckpointManager::HasCkptFile() {
+ Result<bool> exists = file_manager_->CheckpointExists();
+ if (!exists.ok()) {
+ return
::lumina::core::Result<bool>::Err(PaimonToLuminaStatus(exists.status()));
+ }
+ return ::lumina::core::Result<bool>::Ok(exists.value());
+}
+
+std::unique_ptr<::lumina::io::FileReader>
LuminaCheckpointManager::GetCkptFileReader() {
+ Result<std::unique_ptr<InputStream>> input =
file_manager_->OpenCheckpointInputStream();
+ if (!input.ok()) {
+ return nullptr;
+ }
+ std::shared_ptr<InputStream> shared_input = std::move(input).value();
+ return std::make_unique<LuminaFileReader>(shared_input);
+}
+
+} // namespace paimon::lumina
diff --git a/src/paimon/global_index/lumina/lumina_checkpoint_manager.h
b/src/paimon/global_index/lumina/lumina_checkpoint_manager.h
new file mode 100644
index 00000000..e52316e2
--- /dev/null
+++ b/src/paimon/global_index/lumina/lumina_checkpoint_manager.h
@@ -0,0 +1,45 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "lumina/extensions/experimental/CkptManager.h"
+#include "paimon/global_index/io/global_index_checkpoint_file_manager.h"
+
+namespace paimon::lumina {
+
+/// Adapts Paimon's task-scoped global index checkpoint storage to Lumina.
+class LuminaCheckpointManager final : public
::lumina::extensions::experimental::CkptManager {
+ public:
+ explicit LuminaCheckpointManager(
+ const std::shared_ptr<GlobalIndexCheckpointFileManager>& file_manager)
+ : file_manager_(file_manager) {}
+
+ std::unique_ptr<::lumina::io::FileWriter> CreateCkptFileWriter() override;
+
+ ::lumina::core::Result<bool> HasCkptFile() override;
+
+ std::unique_ptr<::lumina::io::FileReader> GetCkptFileReader() override;
+
+ private:
+ std::shared_ptr<GlobalIndexCheckpointFileManager> file_manager_;
+};
+
+} // namespace paimon::lumina
diff --git a/src/paimon/global_index/lumina/lumina_global_index.cpp
b/src/paimon/global_index/lumina/lumina_global_index.cpp
index 98585f21..06383e9f 100644
--- a/src/paimon/global_index/lumina/lumina_global_index.cpp
+++ b/src/paimon/global_index/lumina/lumina_global_index.cpp
@@ -21,25 +21,25 @@
#include <cstring>
#include <numeric>
#include <type_traits>
+#include <unordered_map>
#include <unordered_set>
#include <utility>
#include "arrow/c/bridge.h"
-#include "arrow/c/helpers.h"
+#include "glog/logging.h"
#include "lumina/api/Dataset.h"
#include "lumina/api/LuminaBuilder.h"
#include "lumina/api/LuminaSearcher.h"
#include "lumina/api/OptionsNormalize.h"
#include "lumina/core/Constants.h"
-#include "lumina/core/Status.h"
-#include "lumina/core/Types.h"
#include "lumina/extensions/experimental/BuildCombinedExtensionV0.h"
#include "paimon/common/global_index/global_index_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/common/utils/options_utils.h"
#include "paimon/common/utils/rapidjson_util.h"
-#include "paimon/common/utils/string_utils.h"
#include "paimon/global_index/bitmap_scored_global_index_result.h"
+#include "paimon/global_index/io/global_index_checkpoint_file_manager.h"
+#include "paimon/global_index/lumina/lumina_checkpoint_manager.h"
#include "paimon/global_index/lumina/lumina_file_reader.h"
#include "paimon/global_index/lumina/lumina_file_writer.h"
#include "paimon/global_index/lumina/lumina_utils.h"
@@ -360,6 +360,11 @@ Result<TagValues> LiteralsToTagValues(const
std::vector<Literal>& literals) {
}
}
+bool IsCheckpointEnabled(const std::map<std::string, std::string>&
lumina_options) {
+ return
lumina_options.count(std::string(::lumina::core::kExtensionCkptThreshold)) != 0
||
+
lumina_options.count(std::string(::lumina::core::kExtensionCkptCount)) != 0;
+}
+
} // namespace
Result<std::vector<TagDimensionData>>
LuminaIndexWriter::ExtractTagDataForSegment(
@@ -582,10 +587,19 @@ Result<std::shared_ptr<GlobalIndexWriter>>
LuminaGlobalIndex::CreateWriter(
::lumina::api::BuilderOptions builder_options,
::lumina::api::NormalizeBuilderOptions(std::unordered_map<std::string,
std::string>(
lumina_options.begin(), lumina_options.end())));
+ std::shared_ptr<GlobalIndexCheckpointFileManager> checkpoint_file_manager;
+ if (IsCheckpointEnabled(lumina_options)) {
+ checkpoint_file_manager =
+
std::dynamic_pointer_cast<GlobalIndexCheckpointFileManager>(file_writer);
+ if (!checkpoint_file_manager ||
!checkpoint_file_manager->SupportsCheckpoint()) {
+ return Status::Invalid("Lumina checkpoint requires a
checkpoint-capable file writer");
+ }
+ }
auto lumina_pool = std::make_shared<LuminaMemoryPool>(pool);
return std::make_shared<LuminaIndexWriter>(
field_name, arrow_type, dimension, file_writer,
std::move(builder_options),
- ::lumina::api::IOOptions(), lumina_options, std::move(tag_fields),
lumina_pool);
+ ::lumina::api::IOOptions(), lumina_options, std::move(tag_fields),
checkpoint_file_manager,
+ lumina_pool);
}
Result<LuminaIndexReader::IndexInfo> LuminaIndexReader::GetIndexInfo(
@@ -804,12 +818,25 @@ class LuminaDatasetWithTag : public
::lumina::extensions::experimental::DatasetW
size_t cursor_ = 0;
};
+struct LuminaBuildContext {
+ explicit LuminaBuildContext(::lumina::api::LuminaBuilder&& value) :
builder(std::move(value)) {}
+
+ ::lumina::api::LuminaBuilder builder;
+
std::unique_ptr<::lumina::extensions::experimental::BuildWithCheckpointExtension>
+ checkpoint_extension;
+ std::unique_ptr<::lumina::extensions::experimental::BuildWithTagExtension>
tag_extension;
+
std::unique_ptr<::lumina::extensions::experimental::BuildWithCkptAndTagExtension>
+ checkpoint_tag_extension;
+};
+
LuminaIndexWriter::LuminaIndexWriter(
const std::string& field_name, const std::shared_ptr<arrow::DataType>&
arrow_type,
uint32_t dimension, const std::shared_ptr<GlobalIndexFileWriter>&
file_manager,
::lumina::api::BuilderOptions&& builder_options,
::lumina::api::IOOptions&& io_options,
const std::map<std::string, std::string>& lumina_options,
- std::vector<LuminaTagField>&& tag_fields, const
std::shared_ptr<LuminaMemoryPool>& pool)
+ std::vector<LuminaTagField>&& tag_fields,
+ const std::shared_ptr<GlobalIndexCheckpointFileManager>&
checkpoint_file_manager,
+ const std::shared_ptr<LuminaMemoryPool>& pool)
: pool_(pool),
field_name_(field_name),
arrow_type_(arrow_type),
@@ -818,7 +845,8 @@ LuminaIndexWriter::LuminaIndexWriter(
builder_options_(std::move(builder_options)),
io_options_(std::move(io_options)),
lumina_options_(lumina_options),
- tag_fields_(std::move(tag_fields)) {}
+ tag_fields_(std::move(tag_fields)),
+ checkpoint_file_manager_(checkpoint_file_manager) {}
Status LuminaIndexWriter::AddBatch(::ArrowArray* arrow_array,
std::vector<int64_t>&& relative_row_ids) {
@@ -846,7 +874,6 @@ Status LuminaIndexWriter::AddBatch(::ArrowArray*
arrow_array,
for (int64_t i = 0; i <= field_length; i++) {
bool is_null = (i < field_length) && list_field_array->IsNull(i);
bool is_end = (i == field_length);
-
if (!is_null && !is_end && segment_start == -1) {
segment_start = i;
}
@@ -896,36 +923,89 @@ Result<std::vector<GlobalIndexIOMeta>>
LuminaIndexWriter::Finish() {
if (indexed_count_ == 0) {
return std::vector<GlobalIndexIOMeta>();
}
- ::lumina::core::MemoryResourceConfig memory_resource(pool_.get());
- PAIMON_ASSIGN_OR_RAISE_FROM_LUMINA(
- ::lumina::api::LuminaBuilder builder,
- ::lumina::api::LuminaBuilder::Create(builder_options_,
memory_resource));
- // pretrain
- LuminaDataset dataset1(indexed_count_, dimension_, array_vec_,
array_start_ids_);
- PAIMON_RETURN_NOT_OK_FROM_LUMINA(builder.PretrainFrom(dataset1));
-
- // insert data
- if (tag_fields_.empty()) {
- LuminaDataset dataset2(indexed_count_, dimension_, array_vec_,
array_start_ids_);
- std::vector<std::shared_ptr<arrow::FloatArray>>().swap(array_vec_);
- PAIMON_RETURN_NOT_OK_FROM_LUMINA(builder.InsertFrom(dataset2));
- } else {
- ::lumina::extensions::experimental::BuildWithTagExtension
tag_extension;
- PAIMON_RETURN_NOT_OK_FROM_LUMINA(builder.Attach(tag_extension));
- LuminaDatasetWithTag dataset2(indexed_count_, dimension_, array_vec_,
array_start_ids_,
- tag_data_vec_);
- std::vector<std::shared_ptr<arrow::FloatArray>>().swap(array_vec_);
- std::vector<std::vector<TagDimensionData>>().swap(tag_data_vec_);
-
PAIMON_RETURN_NOT_OK_FROM_LUMINA(tag_extension.InsertFromWithTag(dataset2));
+
+ bool had_checkpoint = false;
+ if (checkpoint_file_manager_) {
+ PAIMON_ASSIGN_OR_RAISE(had_checkpoint,
checkpoint_file_manager_->CheckpointExists());
}
+ auto create_build_context = [&]() ->
Result<std::unique_ptr<LuminaBuildContext>> {
+ ::lumina::core::MemoryResourceConfig memory_resource(pool_.get());
+ PAIMON_ASSIGN_OR_RAISE_FROM_LUMINA(
+ ::lumina::api::LuminaBuilder builder,
+ ::lumina::api::LuminaBuilder::Create(builder_options_,
memory_resource));
+ auto context =
std::make_unique<LuminaBuildContext>(std::move(builder));
+ if (checkpoint_file_manager_) {
+ auto checkpoint_manager =
+
std::make_unique<LuminaCheckpointManager>(checkpoint_file_manager_);
+ auto attach_checkpoint = [&](auto* extension) -> Status {
+
PAIMON_RETURN_NOT_OK_FROM_LUMINA(context->builder.Attach(*extension));
+ PAIMON_RETURN_NOT_OK_FROM_LUMINA(
+ extension->LoadCkptManager(std::move(checkpoint_manager)));
+ return Status::OK();
+ };
+ if (tag_fields_.empty()) {
+ context->checkpoint_extension = std::make_unique<
+
::lumina::extensions::experimental::BuildWithCheckpointExtension>();
+
PAIMON_RETURN_NOT_OK(attach_checkpoint(context->checkpoint_extension.get()));
+ } else {
+ context->checkpoint_tag_extension = std::make_unique<
+
::lumina::extensions::experimental::BuildWithCkptAndTagExtension>();
+
PAIMON_RETURN_NOT_OK(attach_checkpoint(context->checkpoint_tag_extension.get()));
+ }
+ } else if (!tag_fields_.empty()) {
+ context->tag_extension =
+
std::make_unique<::lumina::extensions::experimental::BuildWithTagExtension>();
+
PAIMON_RETURN_NOT_OK_FROM_LUMINA(context->builder.Attach(*context->tag_extension));
+ }
+ return context;
+ };
+
+ auto build_index = [&]() -> Result<std::unique_ptr<LuminaBuildContext>> {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<LuminaBuildContext> context,
create_build_context());
+ // pretrain
+ LuminaDataset pretrain_dataset(indexed_count_, dimension_, array_vec_,
array_start_ids_);
+
PAIMON_RETURN_NOT_OK_FROM_LUMINA(context->builder.PretrainFrom(pretrain_dataset));
+ // insert data
+ if (tag_fields_.empty()) {
+ LuminaDataset insert_dataset(indexed_count_, dimension_,
array_vec_, array_start_ids_);
+
PAIMON_RETURN_NOT_OK_FROM_LUMINA(context->builder.InsertFrom(insert_dataset));
+ } else {
+ LuminaDatasetWithTag insert_dataset(indexed_count_, dimension_,
array_vec_,
+ array_start_ids_,
tag_data_vec_);
+ if (context->checkpoint_tag_extension) {
+ PAIMON_RETURN_NOT_OK_FROM_LUMINA(
+
context->checkpoint_tag_extension->InsertFromWithTag(insert_dataset));
+ } else {
+ PAIMON_RETURN_NOT_OK_FROM_LUMINA(
+ context->tag_extension->InsertFromWithTag(insert_dataset));
+ }
+ }
+ return context;
+ };
+
+ Result<std::unique_ptr<LuminaBuildContext>> build_result = build_index();
+ if (!build_result.ok() && had_checkpoint) {
+ LOG(WARNING) << "Failed to build Lumina index with checkpoint, discard
it and rebuild "
+ "from scratch: "
+ << build_result.status().ToString();
+ PAIMON_RETURN_NOT_OK(checkpoint_file_manager_->DeleteCheckpoint());
+ build_result = build_index();
+ }
+ if (!build_result.ok()) {
+ return build_result.status();
+ }
+ std::unique_ptr<LuminaBuildContext> build_context =
std::move(build_result).value();
+
// dump index
PAIMON_ASSIGN_OR_RAISE(std::string index_file_name,
file_manager_->NewFileName(LuminaDefines::kIdentifier));
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<OutputStream> out,
file_manager_->NewOutputStream(index_file_name));
auto file_writer = std::make_unique<LuminaFileWriter>(out);
- PAIMON_RETURN_NOT_OK_FROM_LUMINA(builder.Dump(std::move(file_writer),
io_options_));
+ PAIMON_RETURN_NOT_OK_FROM_LUMINA(
+ build_context->builder.Dump(std::move(file_writer), io_options_));
+
// prepare GlobalIndexIOMeta
PAIMON_ASSIGN_OR_RAISE(int64_t file_size,
file_manager_->GetFileSize(index_file_name));
std::string options_json;
@@ -933,12 +1013,18 @@ Result<std::vector<GlobalIndexIOMeta>>
LuminaIndexWriter::Finish() {
auto meta_bytes = std::make_shared<Bytes>(options_json,
pool_->GetPaimonPool().get());
GlobalIndexIOMeta meta(file_manager_->ToPath(index_file_name), file_size,
/*metadata=*/meta_bytes);
+ if (checkpoint_file_manager_) {
+ Status status = checkpoint_file_manager_->DeleteCheckpoint();
+ if (!status.ok()) {
+ LOG(WARNING) << "Failed to delete Lumina checkpoints after
successful build: "
+ << status.ToString();
+ }
+ }
return std::vector<GlobalIndexIOMeta>({meta});
}
LuminaIndexReader::LuminaIndexReader(
- const LuminaIndexReader::IndexInfo& index_info,
- std::unique_ptr<::lumina::api::LuminaSearcher>&& searcher,
+ const IndexInfo& index_info,
std::unique_ptr<::lumina::api::LuminaSearcher>&& searcher,
std::unique_ptr<::lumina::extensions::SearchWithFilterExtension>&&
searcher_with_filter,
std::unique_ptr<::lumina::extensions::experimental::SearchWithTagExtension>&&
searcher_with_tag,
const std::shared_ptr<LuminaMemoryPool>& pool)
diff --git a/src/paimon/global_index/lumina/lumina_global_index.h
b/src/paimon/global_index/lumina/lumina_global_index.h
index c2c30475..b4787374 100644
--- a/src/paimon/global_index/lumina/lumina_global_index.h
+++ b/src/paimon/global_index/lumina/lumina_global_index.h
@@ -18,12 +18,11 @@
#pragma once
+#include <cstdint>
#include <map>
#include <memory>
#include <optional>
#include <string>
-#include <unordered_map>
-#include <utility>
#include <vector>
#include "arrow/api.h"
@@ -33,8 +32,12 @@
#include "lumina/extensions/experimental/DatasetWithTag.h"
#include "lumina/extensions/experimental/SearchWithTagExtension.h"
#include "lumina/extensions/experimental/TagFilter.h"
-#include "paimon/global_index/bitmap_global_index_result.h"
+#include "paimon/global_index/global_index_io_meta.h"
+#include "paimon/global_index/global_index_reader.h"
+#include "paimon/global_index/global_index_writer.h"
#include "paimon/global_index/global_indexer.h"
+#include "paimon/global_index/io/global_index_checkpoint_file_manager.h"
+#include "paimon/global_index/io/global_index_file_writer.h"
#include "paimon/global_index/lumina/lumina_memory_pool.h"
#include "paimon/global_index/lumina/lumina_utils.h"
@@ -73,6 +76,8 @@ struct LuminaTagField {
/// lumina.diskann.build.thread_count:64
/// lumina.diskann.build.ef_construction:1024
/// lumina.diskann.build.neighbor_count:64
+/// lumina.extension.build.ckpt.threshold:10000
+/// lumina.extension.build.ckpt.count:3
///
/// - **Index Reader:**
/// No configuration required at load time — settings are stored in
the index metadata,
@@ -87,8 +92,14 @@ class LuminaGlobalIndex : public GlobalIndexer {
explicit LuminaGlobalIndex(const std::map<std::string, std::string>&
options)
: options_(options) {}
+ bool SupportsCheckpoint() const override {
+ return true;
+ }
+
Result<std::optional<std::vector<std::string>>> GetExtraFieldNames() const
override;
+ /// With checkpoints enabled, file_writer must implement
GlobalIndexCheckpointFileManager and
+ /// return true from SupportsCheckpoint().
Result<std::shared_ptr<GlobalIndexWriter>> CreateWriter(
const std::string& field_name, ::ArrowSchema* arrow_schema,
const std::shared_ptr<GlobalIndexFileWriter>& file_writer,
@@ -111,14 +122,14 @@ class LuminaGlobalIndex : public GlobalIndexer {
class LuminaIndexWriter : public GlobalIndexWriter {
public:
- LuminaIndexWriter(const std::string& field_name,
- const std::shared_ptr<arrow::DataType>& arrow_type,
uint32_t dimension,
- const std::shared_ptr<GlobalIndexFileWriter>&
file_manager,
- ::lumina::api::BuilderOptions&& builder_options,
- ::lumina::api::IOOptions&& io_options,
- const std::map<std::string, std::string>& lumina_options,
- std::vector<LuminaTagField>&& tag_fields,
- const std::shared_ptr<LuminaMemoryPool>& pool);
+ LuminaIndexWriter(
+ const std::string& field_name, const std::shared_ptr<arrow::DataType>&
arrow_type,
+ uint32_t dimension, const std::shared_ptr<GlobalIndexFileWriter>&
file_manager,
+ ::lumina::api::BuilderOptions&& builder_options,
::lumina::api::IOOptions&& io_options,
+ const std::map<std::string, std::string>& lumina_options,
+ std::vector<LuminaTagField>&& tag_fields,
+ const std::shared_ptr<GlobalIndexCheckpointFileManager>&
checkpoint_file_manager,
+ const std::shared_ptr<LuminaMemoryPool>& pool);
Status AddBatch(::ArrowArray* arrow_array, std::vector<int64_t>&&
relative_row_ids) override;
@@ -141,6 +152,7 @@ class LuminaIndexWriter : public GlobalIndexWriter {
::lumina::api::IOOptions io_options_;
std::map<std::string, std::string> lumina_options_;
std::vector<LuminaTagField> tag_fields_;
+ std::shared_ptr<GlobalIndexCheckpointFileManager> checkpoint_file_manager_;
std::vector<std::shared_ptr<arrow::FloatArray>> array_vec_;
std::vector<int64_t> array_start_ids_;
std::vector<std::vector<::lumina::extensions::experimental::TagDimensionData>>
tag_data_vec_;
@@ -249,7 +261,7 @@ class LuminaIndexReader : public GlobalIndexReader {
static Result<::lumina::extensions::experimental::TagFilter>
PredicateToTagFilter(
const std::shared_ptr<Predicate>& predicate);
- LuminaIndexReader::IndexInfo index_info_;
+ IndexInfo index_info_;
std::shared_ptr<LuminaMemoryPool> pool_;
std::unique_ptr<::lumina::api::LuminaSearcher> searcher_;
std::unique_ptr<::lumina::extensions::SearchWithFilterExtension>
searcher_with_filter_;
diff --git a/src/paimon/global_index/lumina/lumina_global_index_test.cpp
b/src/paimon/global_index/lumina/lumina_global_index_test.cpp
index 2d54950a..0ad49a96 100644
--- a/src/paimon/global_index/lumina/lumina_global_index_test.cpp
+++ b/src/paimon/global_index/lumina/lumina_global_index_test.cpp
@@ -23,15 +23,23 @@
#include "arrow/c/bridge.h"
#include "arrow/ipc/api.h"
#include "gtest/gtest.h"
+#include "lumina/api/Dataset.h"
+#include "lumina/api/LuminaBuilder.h"
+#include "lumina/core/Constants.h"
+#include "lumina/extensions/experimental/BuildCombinedExtensionV0.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/date_time_utils.h"
#include "paimon/common/utils/path_util.h"
#include "paimon/common/utils/string_utils.h"
#include "paimon/core/global_index/global_index_file_manager.h"
+#include "paimon/core/index/index_checkpoint_path_factory.h"
#include "paimon/core/index/index_path_factory.h"
+#include "paimon/core/utils/file_store_path_factory.h"
#include "paimon/fs/local/local_file_system.h"
#include "paimon/global_index/bitmap_scored_global_index_result.h"
#include "paimon/global_index/global_index_result.h"
+#include "paimon/global_index/lumina/lumina_checkpoint_manager.h"
+#include "paimon/global_index/lumina/lumina_memory_pool.h"
#include "paimon/predicate/predicate_builder.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::lumina::test {
@@ -62,6 +70,63 @@ class LuminaGlobalIndexTest : public ::testing::Test {
std::string index_path_;
};
+ class CountingCheckpointFileManager : public GlobalIndexFileManager {
+ public:
+ using GlobalIndexFileManager::GlobalIndexFileManager;
+
+ Result<bool> CheckpointExists() const override {
+ check_checkpoint_count_++;
+ return GlobalIndexFileManager::CheckpointExists();
+ }
+
+ Result<std::unique_ptr<OutputStream>> CreateCheckpointOutputStream()
const override {
+ create_checkpoint_count_++;
+ return GlobalIndexFileManager::CreateCheckpointOutputStream();
+ }
+
+ Result<std::unique_ptr<InputStream>> OpenCheckpointInputStream() const
override {
+ open_checkpoint_count_++;
+ return GlobalIndexFileManager::OpenCheckpointInputStream();
+ }
+
+ mutable int32_t check_checkpoint_count_ = 0;
+ mutable int32_t create_checkpoint_count_ = 0;
+ mutable int32_t open_checkpoint_count_ = 0;
+ };
+
+ class TestLuminaDataset : public ::lumina::api::Dataset {
+ public:
+ TestLuminaDataset(uint32_t dimension, const std::vector<float>&
vectors,
+ const std::vector<::lumina::core::vector_id_t>& ids)
+ : dimension_(dimension), vectors_(vectors), ids_(ids) {}
+
+ uint32_t Dim() const noexcept override {
+ return dimension_;
+ }
+
+ uint64_t TotalSize() const noexcept override {
+ return ids_.size();
+ }
+
+ ::lumina::core::Result<uint64_t> GetNextBatch(
+ std::vector<float>& vector_buffer,
+ std::vector<::lumina::core::vector_id_t>& id_buffer) noexcept
override {
+ if (consumed_) {
+ return ::lumina::core::Result<uint64_t>::Ok(0);
+ }
+ vector_buffer = vectors_;
+ id_buffer = ids_;
+ consumed_ = true;
+ return ::lumina::core::Result<uint64_t>::Ok(ids_.size());
+ }
+
+ private:
+ uint32_t dimension_;
+ std::vector<float> vectors_;
+ std::vector<::lumina::core::vector_id_t> ids_;
+ bool consumed_ = false;
+ };
+
std::unique_ptr<::ArrowSchema> CreateArrowSchema(
const std::shared_ptr<arrow::DataType>& data_type) const {
auto c_schema = std::make_unique<::ArrowSchema>();
@@ -69,14 +134,47 @@ class LuminaGlobalIndexTest : public ::testing::Test {
return c_schema;
}
+ Result<std::shared_ptr<FileStorePathFactory>> CreateFileStorePathFactory(
+ const std::string& table_path) const {
+ return FileStorePathFactory::Create(
+ table_path, arrow::schema({}), /*partition_keys=*/{},
/*default_part_value=*/"",
+ /*identifier=*/"mock", /*data_file_prefix=*/"data-",
+ /*legacy_partition_name_enabled=*/true, /*external_paths=*/{},
+ /*global_index_external_path=*/std::nullopt,
+ /*index_file_in_data_file_dir=*/false, pool_);
+ }
+
+ Result<std::unique_ptr<IndexCheckpointPathFactory>>
CreateCheckpointPathFactory(
+ const std::string& table_path, const std::string& index_type, const
std::string& field_name,
+ const Range& range) const {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStorePathFactory>
file_store_path_factory,
+ CreateFileStorePathFactory(table_path));
+ return file_store_path_factory->CreateGlobalIndexCheckpointPathFactory(
+ index_type, field_name, range, "task-1");
+ }
+
+ Result<std::shared_ptr<CountingCheckpointFileManager>>
CreateCountingCheckpointFileManager(
+ const std::shared_ptr<FileStorePathFactory>& path_factory, const
Range& range) const {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_path_factory,
+ path_factory->CreateGlobalIndexCheckpointPathFactory("lumina",
"f0", range, "task-1"));
+ return std::make_shared<CountingCheckpointFileManager>(
+ fs_, path_factory->CreateGlobalIndexFileFactory(),
std::move(checkpoint_path_factory));
+ }
+
Result<GlobalIndexIOMeta> WriteGlobalIndex(const std::string& index_root,
const
std::shared_ptr<arrow::DataType>& data_type,
const std::map<std::string,
std::string>& options,
const
std::shared_ptr<arrow::Array>& array,
const Range& expected_range)
const {
auto global_index = std::make_shared<LuminaGlobalIndex>(options);
- auto path_factory = std::make_shared<FakeIndexPathFactory>(index_root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStorePathFactory>
path_factory,
+ CreateFileStorePathFactory(index_root));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_path_factory,
+
path_factory->CreateGlobalIndexCheckpointPathFactory(
+ "lumina", "f0", expected_range, "task-1"));
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory->CreateGlobalIndexFileFactory(),
std::move(checkpoint_path_factory));
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexWriter>
global_writer,
global_index->CreateWriter("f0",
CreateArrowSchema(data_type).get(),
@@ -127,7 +225,8 @@ class LuminaGlobalIndexTest : public ::testing::Test {
const std::map<std::string, std::string>& options, const
GlobalIndexIOMeta& meta) const {
auto global_index = std::make_shared<LuminaGlobalIndex>(options);
auto path_factory = std::make_shared<FakeIndexPathFactory>(index_root);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
return global_index->CreateReader(CreateArrowSchema(data_type).get(),
file_reader, {meta},
pool_);
}
@@ -252,12 +351,344 @@ TEST_F(LuminaGlobalIndexTest, TestWithFilter) {
}
}
+TEST_F(LuminaGlobalIndexTest, TestCheckpointCapabilityOnlyRequiredWhenEnabled)
{
+ ASSERT_TRUE(LuminaGlobalIndex(options_).SupportsCheckpoint());
+ auto dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> path_factory,
+ CreateFileStorePathFactory(dir->Str()));
+ auto plain_manager =
+ std::make_shared<GlobalIndexFileManager>(fs_,
path_factory->CreateGlobalIndexFileFactory(),
+
/*checkpoint_path_factory=*/nullptr);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<CountingCheckpointFileManager>
checkpoint_manager,
+ CreateCountingCheckpointFileManager(path_factory,
Range(0, 3)));
+ std::map<std::string, std::string> options = options_;
+ options["other.extension.build.ckpt.count"] = "1";
+ ASSERT_OK(LuminaGlobalIndex(options).CreateWriter("f0",
CreateArrowSchema(data_type_).get(),
+ plain_manager, pool_));
+ ASSERT_OK(LuminaGlobalIndex(options).CreateWriter("f0",
CreateArrowSchema(data_type_).get(),
+ checkpoint_manager,
pool_));
+
+ for (const std::string& checkpoint_key :
+ std::vector<std::string>{"extension.build.ckpt.count",
"extension.build.ckpt.threshold"}) {
+ auto enabled = options;
+ enabled["lumina." + checkpoint_key] = "1";
+ ASSERT_NOK_WITH_MSG(LuminaGlobalIndex(enabled).CreateWriter(
+ "f0", CreateArrowSchema(data_type_).get(),
plain_manager, pool_),
+ "Lumina checkpoint requires a checkpoint-capable
file writer");
+ ASSERT_NOK_WITH_MSG(LuminaGlobalIndex(enabled).CreateWriter(
+ "f0", CreateArrowSchema(data_type_).get(),
nullptr, pool_),
+ "Lumina checkpoint requires a checkpoint-capable
file writer");
+ ASSERT_OK(LuminaGlobalIndex(enabled).CreateWriter("f0",
CreateArrowSchema(data_type_).get(),
+ checkpoint_manager,
pool_));
+ }
+ ASSERT_EQ(checkpoint_manager->check_checkpoint_count_, 0);
+ ASSERT_EQ(checkpoint_manager->create_checkpoint_count_, 0);
+}
+
+TEST_F(LuminaGlobalIndexTest, TestBuildWithCheckpoint) {
+ auto test_root_dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(test_root_dir);
+ std::string test_root = test_root_dir->Str();
+
+ std::map<std::string, std::string> checkpoint_options = options_;
+ checkpoint_options["lumina.extension.build.ckpt.threshold"] = "1";
+ checkpoint_options["lumina.extension.build.ckpt.count"] = "1";
+
+ auto global_index =
std::make_shared<LuminaGlobalIndex>(checkpoint_options);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> path_factory,
+ CreateFileStorePathFactory(test_root));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<CountingCheckpointFileManager>
file_manager,
+ CreateCountingCheckpointFileManager(path_factory,
Range(0, 3)));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexWriter> global_writer,
+ global_index->CreateWriter("f0", CreateArrowSchema(data_type_).get(),
file_manager, pool_));
+
+ ArrowArray c_array;
+ ASSERT_TRUE(arrow::ExportArray(*array_, &c_array).ok());
+ ASSERT_OK(global_writer->AddBatch(&c_array, {0, 1, 2, 3}));
+ ASSERT_OK_AND_ASSIGN(std::vector<GlobalIndexIOMeta> result_metas,
global_writer->Finish());
+ ASSERT_EQ(result_metas.size(), 1);
+ ASSERT_GT(file_manager->create_checkpoint_count_, 0);
+ ASSERT_OK_AND_ASSIGN(bool checkpoint_exists,
file_manager->CheckpointExists());
+ ASSERT_FALSE(checkpoint_exists);
+
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexReader> reader,
+ CreateGlobalIndexReader(test_root, data_type_, checkpoint_options,
result_metas[0]));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult>
scored_result,
+
reader->VisitVectorSearch(std::make_shared<VectorSearch>(
+ /*field_name=*/"f0", /*limit=*/4, query_,
/*filter=*/nullptr,
+ /*predicate=*/nullptr,
/*distance_type=*/std::nullopt,
+ /*options=*/checkpoint_options)));
+ CheckResult(scored_result, {3l, 1l, 2l, 0l}, {0.01f, 2.01f, 2.21f, 4.21f});
+}
+
+TEST_F(LuminaGlobalIndexTest, TestCheckpointFileManagement) {
+ auto test_root_dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(test_root_dir);
+ std::string test_root = test_root_dir->Str();
+ std::string prefix = "lumina-global-index-f0-10-20-task-1-";
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_path_factory,
+ CreateCheckpointPathFactory(test_root, "lumina",
"f0", Range(10, 20)));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> path_factory,
+ CreateFileStorePathFactory(test_root));
+ auto file_manager = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory->CreateGlobalIndexFileFactory(),
std::move(checkpoint_path_factory));
+
+ ASSERT_OK_AND_ASSIGN(bool checkpoint_exists,
file_manager->CheckpointExists());
+ ASSERT_FALSE(checkpoint_exists);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> first_output,
+ file_manager->CreateCheckpointOutputStream());
+ ASSERT_OK_AND_ASSIGN(std::string first_path, first_output->GetUri());
+ ASSERT_TRUE(StringUtils::StartsWith(PathUtil::GetName(first_path),
prefix));
+ ASSERT_TRUE(StringUtils::EndsWith(first_path, "-0.index.ckpt"));
+ ASSERT_OK_AND_ASSIGN(int64_t first_written, first_output->Write("old", 3));
+ ASSERT_EQ(first_written, 3);
+ ASSERT_OK(first_output->Close());
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> second_output,
+ file_manager->CreateCheckpointOutputStream());
+ ASSERT_OK_AND_ASSIGN(std::string second_path, second_output->GetUri());
+ ASSERT_TRUE(StringUtils::StartsWith(PathUtil::GetName(second_path),
prefix));
+ ASSERT_TRUE(StringUtils::EndsWith(second_path, "-1.index.ckpt"));
+ ASSERT_OK_AND_ASSIGN(int64_t second_written,
second_output->Write("latest", 6));
+ ASSERT_EQ(second_written, 6);
+ ASSERT_OK(second_output->Close());
+
+ std::string checkpoint_dir = PathUtil::JoinPath(test_root,
"index/checkpoint");
+ ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(checkpoint_dir, prefix +
"9.index.ckpt"), "id-nine",
+ /*overwrite=*/false));
+ ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(checkpoint_dir, prefix +
"10.index.ckpt"), "id-ten",
+ /*overwrite=*/false));
+ std::string unrelated_file =
+ PathUtil::JoinPath(checkpoint_dir,
"lumina-global-index-other-10-20-task-2-99.index.ckpt");
+ ASSERT_OK(fs_->WriteFile(unrelated_file, "unrelated",
/*overwrite=*/false));
+ std::string malformed_file =
+ PathUtil::JoinPath(checkpoint_dir, prefix + "invalid-99.index.ckpt");
+ ASSERT_OK(fs_->WriteFile(malformed_file, "malformed",
/*overwrite=*/false));
+
+ ASSERT_OK_AND_ASSIGN(checkpoint_path_factory,
+ CreateCheckpointPathFactory(test_root, "lumina",
"f0", Range(10, 20)));
+ file_manager = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory->CreateGlobalIndexFileFactory(),
std::move(checkpoint_path_factory));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> latest_input,
+ file_manager->OpenCheckpointInputStream());
+ char buffer[6];
+ ASSERT_OK_AND_ASSIGN(int64_t read_bytes, latest_input->Read(buffer,
sizeof(buffer)));
+ ASSERT_EQ(read_bytes, 6);
+ ASSERT_EQ(std::string(buffer, sizeof(buffer)), "id-ten");
+ ASSERT_OK(latest_input->Close());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> next_output,
+ file_manager->CreateCheckpointOutputStream());
+ ASSERT_OK_AND_ASSIGN(std::string next_path, next_output->GetUri());
+ ASSERT_TRUE(StringUtils::EndsWith(next_path, "-11.index.ckpt"));
+ ASSERT_OK(next_output->Close());
+
+ ASSERT_OK(file_manager->DeleteCheckpoint());
+ ASSERT_OK_AND_ASSIGN(checkpoint_exists, file_manager->CheckpointExists());
+ ASSERT_FALSE(checkpoint_exists);
+ ASSERT_OK_AND_ASSIGN(bool unrelated_exists, fs_->Exists(unrelated_file));
+ ASSERT_TRUE(unrelated_exists);
+ ASSERT_OK_AND_ASSIGN(bool malformed_exists, fs_->Exists(malformed_file));
+ ASSERT_TRUE(malformed_exists);
+}
+
+TEST_F(LuminaGlobalIndexTest, TestResumeFromCheckpoint) {
+ auto test_root_dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(test_root_dir);
+ std::string test_root = test_root_dir->Str();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> path_factory,
+ CreateFileStorePathFactory(test_root));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<CountingCheckpointFileManager>
file_manager,
+ CreateCountingCheckpointFileManager(path_factory,
Range(0, 3)));
+
+ std::map<std::string, std::string> checkpoint_options = options_;
+ checkpoint_options["lumina.extension.build.ckpt.threshold"] = "1";
+ checkpoint_options["lumina.extension.build.ckpt.count"] = "1";
+ std::vector<float> vectors = {
+ 0.0f, 0.0f, 0.0f, 0.0f, 0.0f, 1.0f, 0.0f, 1.0f,
+ 1.0f, 0.0f, 1.0f, 0.0f, 1.0f, 1.0f, 1.0f, 1.0f,
+ };
+ std::vector<::lumina::core::vector_id_t> ids = {0, 1, 2, 3};
+
+ {
+ ::lumina::api::BuilderOptions builder_options;
+ builder_options.Set(::lumina::core::kIndexType,
::lumina::core::kIndexTypeBruteforce)
+ .Set(::lumina::core::kDimension, static_cast<int64_t>(4))
+ .Set(::lumina::core::kDistanceMetric, ::lumina::core::kDistanceL2)
+ .Set(::lumina::core::kEncodingType,
::lumina::core::kEncodingRawf32)
+ .Set(::lumina::core::kExtensionCkptThreshold,
static_cast<int64_t>(1))
+ .Set(::lumina::core::kExtensionCkptCount, static_cast<int64_t>(1));
+ LuminaMemoryPool lumina_pool(pool_);
+ ::lumina::core::MemoryResourceConfig memory_resource(&lumina_pool);
+ auto builder_result =
+ ::lumina::api::LuminaBuilder::Create(builder_options,
memory_resource);
+ ASSERT_TRUE(builder_result.IsOk()) <<
builder_result.GetStatus().Message();
+ ::lumina::api::LuminaBuilder builder =
std::move(builder_result).TakeValue();
+ ::lumina::extensions::experimental::BuildWithCheckpointExtension
checkpoint_extension;
+ ASSERT_TRUE(builder.Attach(checkpoint_extension).IsOk());
+ ASSERT_TRUE(checkpoint_extension
+
.LoadCkptManager(std::make_unique<LuminaCheckpointManager>(file_manager))
+ .IsOk());
+ TestLuminaDataset pretrain_dataset(/*dimension=*/4, vectors, ids);
+ ASSERT_TRUE(builder.PretrainFrom(pretrain_dataset).IsOk());
+ TestLuminaDataset insert_dataset(/*dimension=*/4, vectors, ids);
+ ASSERT_TRUE(builder.InsertFrom(insert_dataset).IsOk());
+ }
+ ASSERT_GT(file_manager->create_checkpoint_count_, 0);
+ ASSERT_OK_AND_ASSIGN(bool checkpoint_exists,
file_manager->CheckpointExists());
+ ASSERT_TRUE(checkpoint_exists);
+
+ auto global_index =
std::make_shared<LuminaGlobalIndex>(checkpoint_options);
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexWriter> global_writer,
+ global_index->CreateWriter("f0", CreateArrowSchema(data_type_).get(),
file_manager, pool_));
+ ArrowArray c_array;
+ ASSERT_TRUE(arrow::ExportArray(*array_, &c_array).ok());
+ ASSERT_OK(global_writer->AddBatch(&c_array, {0, 1, 2, 3}));
+ ASSERT_OK_AND_ASSIGN(std::vector<GlobalIndexIOMeta> result_metas,
global_writer->Finish());
+ ASSERT_EQ(result_metas.size(), 1);
+ ASSERT_GT(file_manager->open_checkpoint_count_, 0);
+ ASSERT_OK_AND_ASSIGN(checkpoint_exists, file_manager->CheckpointExists());
+ ASSERT_FALSE(checkpoint_exists);
+
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexReader> reader,
+ CreateGlobalIndexReader(test_root, data_type_, checkpoint_options,
result_metas[0]));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult>
scored_result,
+
reader->VisitVectorSearch(std::make_shared<VectorSearch>(
+ /*field_name=*/"f0", /*limit=*/4, query_,
/*filter=*/nullptr,
+ /*predicate=*/nullptr,
/*distance_type=*/std::nullopt,
+ /*options=*/checkpoint_options)));
+ CheckResult(scored_result, {3l, 1l, 2l, 0l}, {0.01f, 2.01f, 2.21f, 4.21f});
+}
+
+TEST_F(LuminaGlobalIndexTest, TestDiscardInvalidCheckpointAndRebuild) {
+ auto test_root_dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(test_root_dir);
+ std::string test_root = test_root_dir->Str();
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> path_factory,
+ CreateFileStorePathFactory(test_root));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<CountingCheckpointFileManager>
file_manager,
+ CreateCountingCheckpointFileManager(path_factory,
Range(0, 3)));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<OutputStream> invalid_checkpoint,
+ file_manager->CreateCheckpointOutputStream());
+ ASSERT_OK_AND_ASSIGN(int64_t written, invalid_checkpoint->Write("invalid
checkpoint", 18));
+ ASSERT_EQ(written, 18);
+ ASSERT_OK(invalid_checkpoint->Close());
+
+ std::map<std::string, std::string> checkpoint_options = options_;
+ checkpoint_options["lumina.extension.build.ckpt.threshold"] = "1";
+ checkpoint_options["lumina.extension.build.ckpt.count"] = "1";
+ auto global_index =
std::make_shared<LuminaGlobalIndex>(checkpoint_options);
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexWriter> global_writer,
+ global_index->CreateWriter("f0", CreateArrowSchema(data_type_).get(),
file_manager, pool_));
+ ArrowArray c_array;
+ ASSERT_TRUE(arrow::ExportArray(*array_, &c_array).ok());
+ ASSERT_OK(global_writer->AddBatch(&c_array, {0, 1, 2, 3}));
+ ASSERT_OK_AND_ASSIGN(std::vector<GlobalIndexIOMeta> result_metas,
global_writer->Finish());
+ ASSERT_EQ(result_metas.size(), 1);
+ ASSERT_GT(file_manager->open_checkpoint_count_, 0);
+ ASSERT_OK_AND_ASSIGN(bool checkpoint_exists,
file_manager->CheckpointExists());
+ ASSERT_FALSE(checkpoint_exists);
+
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexReader> reader,
+ CreateGlobalIndexReader(test_root, data_type_, checkpoint_options,
result_metas[0]));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult>
scored_result,
+
reader->VisitVectorSearch(std::make_shared<VectorSearch>(
+ /*field_name=*/"f0", /*limit=*/4, query_,
/*filter=*/nullptr,
+ /*predicate=*/nullptr,
/*distance_type=*/std::nullopt,
+ /*options=*/checkpoint_options)));
+ CheckResult(scored_result, {3l, 1l, 2l, 0l}, {0.01f, 2.01f, 2.21f, 4.21f});
+}
+
+TEST_F(LuminaGlobalIndexTest, TestBuildFailureDiscardsCheckpointAndRetries) {
+ class DeletionCountingCheckpointFileManager : public
CountingCheckpointFileManager {
+ public:
+ using CountingCheckpointFileManager::CountingCheckpointFileManager;
+
+ Status DeleteCheckpoint() const override {
+ delete_count_++;
+ return CountingCheckpointFileManager::DeleteCheckpoint();
+ }
+
+ mutable int32_t delete_count_ = 0;
+ };
+
+ auto dir = paimon::test::UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileStorePathFactory> path_factory,
+ CreateFileStorePathFactory(dir->Str()));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexCheckpointPathFactory>
checkpoint_path_factory,
+ path_factory->CreateGlobalIndexCheckpointPathFactory(
+ "lumina", "f0", Range(0, 3), "task-1"));
+ auto file_manager =
std::make_shared<DeletionCountingCheckpointFileManager>(
+ fs_, path_factory->CreateGlobalIndexFileFactory(),
std::move(checkpoint_path_factory));
+
+ ::lumina::api::BuilderOptions builder_options;
+ builder_options.Set(::lumina::core::kIndexType,
::lumina::core::kIndexTypeBruteforce)
+ .Set(::lumina::core::kDimension, static_cast<int64_t>(4))
+ .Set(::lumina::core::kDistanceMetric, ::lumina::core::kDistanceL2)
+ .Set(::lumina::core::kEncodingType, ::lumina::core::kEncodingRawf32)
+ .Set(::lumina::core::kExtensionCkptThreshold, static_cast<int64_t>(1))
+ .Set(::lumina::core::kExtensionCkptCount, static_cast<int64_t>(1));
+ LuminaMemoryPool lumina_pool(pool_);
+ ::lumina::core::MemoryResourceConfig memory_resource(&lumina_pool);
+ auto builder_result =
::lumina::api::LuminaBuilder::Create(builder_options, memory_resource);
+ ASSERT_TRUE(builder_result.IsOk()) << builder_result.GetStatus().Message();
+ {
+ ::lumina::api::LuminaBuilder builder =
std::move(builder_result).TakeValue();
+ ::lumina::extensions::experimental::BuildWithCheckpointExtension
checkpoint_extension;
+ ASSERT_TRUE(builder.Attach(checkpoint_extension).IsOk());
+ ASSERT_TRUE(checkpoint_extension
+
.LoadCkptManager(std::make_unique<LuminaCheckpointManager>(file_manager))
+ .IsOk());
+ std::vector<float> vectors = {
+ 0.0f, 0.0f, 0.0f, 0.0f, 0.0f, 1.0f, 0.0f, 1.0f,
+ 1.0f, 0.0f, 1.0f, 0.0f, 1.0f, 1.0f, 1.0f, 1.0f,
+ };
+ std::vector<::lumina::core::vector_id_t> ids = {0, 1, 2, 3};
+ TestLuminaDataset pretrain_dataset(/*dimension=*/4, vectors, ids);
+ ASSERT_TRUE(builder.PretrainFrom(pretrain_dataset).IsOk());
+ TestLuminaDataset insert_dataset(/*dimension=*/4, vectors, ids);
+ ASSERT_TRUE(builder.InsertFrom(insert_dataset).IsOk());
+ }
+ ASSERT_OK_AND_ASSIGN(bool checkpoint_exists,
file_manager->CheckpointExists());
+ ASSERT_TRUE(checkpoint_exists);
+
+ std::map<std::string, std::string> checkpoint_options = options_;
+ checkpoint_options["lumina.extension.build.ckpt.threshold"] = "1";
+ checkpoint_options["lumina.extension.build.ckpt.count"] = "1";
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<GlobalIndexWriter> writer,
+ LuminaGlobalIndex(checkpoint_options)
+ .CreateWriter("f0", CreateArrowSchema(data_type_).get(),
file_manager, pool_));
+ // The two-row build fails after loading the checkpoint, and must be
retried without it.
+ file_manager->check_checkpoint_count_ = 0;
+ ArrowArray c_array;
+ ASSERT_TRUE(arrow::ExportArray(*array_->Slice(0, 2), &c_array).ok());
+ ASSERT_OK(writer->AddBatch(&c_array, {0, 1}));
+ ASSERT_NOK(writer->Finish());
+ ASSERT_GT(file_manager->open_checkpoint_count_, 0);
+ ASSERT_EQ(file_manager->delete_count_, 1);
+ ASSERT_GE(file_manager->check_checkpoint_count_, 3);
+ ASSERT_OK_AND_ASSIGN(checkpoint_exists, file_manager->CheckpointExists());
+ ASSERT_FALSE(checkpoint_exists);
+}
+
TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithTagFilter) {
auto test_root_dir = paimon::test::UniqueTestDirectory::Create();
ASSERT_TRUE(test_root_dir);
std::string test_root = test_root_dir->Str();
std::map<std::string, std::string> tag_options = options_;
+ tag_options["lumina.extension.build.ckpt.threshold"] = "1";
+ tag_options["lumina.extension.build.ckpt.count"] = "1";
tag_options["lumina.extension.build.tag.tag_schema"] =
R"({"key_name":"color","type":"enum","value_type":"string"})";
@@ -276,6 +707,9 @@ TEST_F(LuminaGlobalIndexTest,
TestWriteAndReadWithTagFilter) {
ASSERT_OK_AND_ASSIGN(
GlobalIndexIOMeta meta,
WriteGlobalIndex(test_root, tag_data_type, tag_options, tag_array,
Range(0, 3)));
+ std::vector<BasicFileStatus> checkpoint_files;
+ ASSERT_OK(fs_->ListDir(PathUtil::JoinPath(test_root, "index/checkpoint"),
&checkpoint_files));
+ ASSERT_TRUE(checkpoint_files.empty());
ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexReader> reader,
CreateGlobalIndexReader(test_root, data_type_,
tag_options, meta));
@@ -787,7 +1221,8 @@ TEST_F(LuminaGlobalIndexTest, TestInvalidInputs) {
{
auto global_index = std::make_shared<LuminaGlobalIndex>(options_);
auto path_factory =
std::make_shared<FakeIndexPathFactory>(index_root);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
ASSERT_NOK_WITH_MSG(global_index->CreateReader(CreateArrowSchema(data_type_).get(),
file_reader, {meta,
meta}, pool_),
@@ -1005,7 +1440,8 @@ TEST_F(LuminaGlobalIndexTest, TestWriteWithAllNullRows) {
auto global_index = std::make_shared<LuminaGlobalIndex>(options_);
auto path_factory = std::make_shared<FakeIndexPathFactory>(test_root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<GlobalIndexWriter> global_writer,
@@ -1077,7 +1513,8 @@ TEST_F(LuminaGlobalIndexTest,
TestWriteWithNullAcrossMultipleBatches) {
auto global_index = std::make_shared<LuminaGlobalIndex>(options_);
auto path_factory = std::make_shared<FakeIndexPathFactory>(test_root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
ASSERT_OK_AND_ASSIGN(
std::shared_ptr<GlobalIndexWriter> global_writer,
diff --git a/src/paimon/global_index/tantivy/tantivy_equivalence_test.cpp
b/src/paimon/global_index/tantivy/tantivy_equivalence_test.cpp
index 44f780ac..07f06200 100644
--- a/src/paimon/global_index/tantivy/tantivy_equivalence_test.cpp
+++ b/src/paimon/global_index/tantivy/tantivy_equivalence_test.cpp
@@ -117,7 +117,8 @@ class TantivyEquivalenceTest : public ::testing::Test {
const std::string& root) {
EXPECT_OK_AND_ASSIGN(auto indexer,
GlobalIndexerFactory::Get(factory_id, options));
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
EXPECT_OK_AND_ASSIGN(
auto writer,
indexer->CreateWriter("f0", CreateArrowSchema(data_type).get(),
file_writer, pool_));
@@ -139,7 +140,8 @@ class TantivyEquivalenceTest : public ::testing::Test {
const std::string& root) {
EXPECT_OK_AND_ASSIGN(auto indexer,
GlobalIndexerFactory::Get(factory_id, options));
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
EXPECT_OK_AND_ASSIGN(auto reader,
indexer->CreateReader(CreateArrowSchema(data_type).get(),
file_reader,
{meta}, pool_));
return reader;
diff --git a/src/paimon/global_index/tantivy/tantivy_filter_limit_test.cpp
b/src/paimon/global_index/tantivy/tantivy_filter_limit_test.cpp
index 5de18683..19df5ac6 100644
--- a/src/paimon/global_index/tantivy/tantivy_filter_limit_test.cpp
+++ b/src/paimon/global_index/tantivy/tantivy_filter_limit_test.cpp
@@ -76,7 +76,8 @@ class TantivyFilterLimitTest : public ::testing::Test {
std::string root = root_dir->Str();
kept_dirs_.push_back(std::move(root_dir));
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto fm = std::make_shared<GlobalIndexFileManager>(fs_, path_factory);
+ auto fm = std::make_shared<GlobalIndexFileManager>(fs_, path_factory,
+
/*checkpoint_path_factory=*/nullptr);
auto data_type = arrow::struct_({arrow::field("f0", arrow::utf8())});
EXPECT_OK_AND_ASSIGN(auto writer_res, TantivyGlobalIndexWriter::Create(
"f0", data_type, fm,
options, GetDefaultPool()));
diff --git a/src/paimon/global_index/tantivy/tantivy_index_test.cpp
b/src/paimon/global_index/tantivy/tantivy_index_test.cpp
index 199e347b..d901d77c 100644
--- a/src/paimon/global_index/tantivy/tantivy_index_test.cpp
+++ b/src/paimon/global_index/tantivy/tantivy_index_test.cpp
@@ -85,7 +85,8 @@ class TantivyGlobalIndexIntegrationTest : public
::testing::Test {
int64_t
/*unused_expected_range_end*/) const {
auto global_index = std::make_shared<TantivyGlobalIndex>(options);
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexWriter> w,
global_index->CreateWriter("f0",
CreateArrowSchema(data_type).get(),
file_writer, pool_));
@@ -111,7 +112,8 @@ class TantivyGlobalIndexIntegrationTest : public
::testing::Test {
const std::map<std::string, std::string>& options, const
GlobalIndexIOMeta& meta) const {
auto global_index = std::make_shared<TantivyGlobalIndex>(options);
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
return global_index->CreateReader(CreateArrowSchema(data_type).get(),
file_reader, {meta},
pool_);
}
diff --git a/src/paimon/global_index/tantivy/tantivy_java_compat_test.cpp
b/src/paimon/global_index/tantivy/tantivy_java_compat_test.cpp
index b998402c..c3358482 100644
--- a/src/paimon/global_index/tantivy/tantivy_java_compat_test.cpp
+++ b/src/paimon/global_index/tantivy/tantivy_java_compat_test.cpp
@@ -96,7 +96,8 @@ class JavaCompatTest : public ::testing::Test {
std::map<std::string, std::string> options;
auto global_index = std::make_shared<TantivyGlobalIndex>(options);
auto path_factory = std::make_shared<FixturePathFactory>(fixture_dir);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
auto data_type = arrow::struct_({arrow::field("f0", arrow::utf8())});
auto c_schema = std::make_unique<::ArrowSchema>();
@@ -417,7 +418,8 @@ TEST_F(JavaCompatTest,
CppWriteDefaultTokenizerForJavaCrossRead) {
auto reader_factory =
std::make_shared<TantivyGlobalIndex>(std::map<std::string,
std::string>{});
auto reader_path_factory = std::make_shared<FixturePathFactory>(out_dir);
- auto reader_file_mgr = std::make_shared<GlobalIndexFileManager>(fs_,
reader_path_factory);
+ auto reader_file_mgr = std::make_shared<GlobalIndexFileManager>(
+ fs_, reader_path_factory, /*checkpoint_path_factory=*/nullptr);
auto c_schema = std::make_unique<::ArrowSchema>();
ASSERT_TRUE(arrow::ExportType(*data_type, c_schema.get()).ok());
diff --git a/src/paimon/global_index/tantivy/tantivy_lucene_coexist_test.cpp
b/src/paimon/global_index/tantivy/tantivy_lucene_coexist_test.cpp
index 2b6fdeb5..e837a0de 100644
--- a/src/paimon/global_index/tantivy/tantivy_lucene_coexist_test.cpp
+++ b/src/paimon/global_index/tantivy/tantivy_lucene_coexist_test.cpp
@@ -101,7 +101,8 @@ class TantivyLuceneCoexistTest : public ::testing::Test {
return Status::Invalid(fmt::format("factory returned null for {}",
impl.factory_id));
}
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
PAIMON_ASSIGN_OR_RAISE(
std::shared_ptr<GlobalIndexWriter> w,
indexer->CreateWriter("f0", CreateArrowSchema(data_type).get(),
file_writer, pool_));
@@ -127,7 +128,8 @@ class TantivyLuceneCoexistTest : public ::testing::Test {
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<GlobalIndexer> indexer,
GlobalIndexerFactory::Get(impl.factory_id,
options));
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
return indexer->CreateReader(CreateArrowSchema(data_type).get(),
file_reader, {meta},
pool_);
}
diff --git a/src/paimon/global_index/tantivy/tantivy_streaming_test.cpp
b/src/paimon/global_index/tantivy/tantivy_streaming_test.cpp
index 36ca7a3a..d5de3265 100644
--- a/src/paimon/global_index/tantivy/tantivy_streaming_test.cpp
+++ b/src/paimon/global_index/tantivy/tantivy_streaming_test.cpp
@@ -106,7 +106,8 @@ class StreamingTestFixture : public ::testing::Test {
EXPECT_TRUE(arrow::ExportType(*data_type, c_schema.get()).ok());
auto global_index = std::make_shared<TantivyGlobalIndex>(options);
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
EXPECT_OK_AND_ASSIGN(auto w,
global_index->CreateWriter("f0", c_schema.get(),
file_writer, pool_));
::ArrowArray c_array;
@@ -132,7 +133,8 @@ class StreamingTestFixture : public ::testing::Test {
EXPECT_TRUE(arrow::ExportType(*data_type, c_schema.get()).ok());
auto global_index = std::make_shared<TantivyGlobalIndex>(options);
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto file_reader = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_reader = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
EXPECT_OK_AND_ASSIGN(
auto reader, global_index->CreateReader(c_schema.get(),
file_reader, {meta}, pool_));
return reader;
diff --git a/src/paimon/global_index/tantivy/tantivy_writer_test.cpp
b/src/paimon/global_index/tantivy/tantivy_writer_test.cpp
index 0d623b88..886f7f5c 100644
--- a/src/paimon/global_index/tantivy/tantivy_writer_test.cpp
+++ b/src/paimon/global_index/tantivy/tantivy_writer_test.cpp
@@ -132,7 +132,8 @@ class TantivyGlobalIndexWriterTest : public ::testing::Test
{
const std::map<std::string, std::string>& options,
const std::shared_ptr<arrow::Array>& array) {
auto path_factory = std::make_shared<FakeIndexPathFactory>(root);
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
PAIMON_ASSIGN_OR_RAISE(auto writer, TantivyGlobalIndexWriter::Create(
"f0", data_type, file_writer,
options, pool_));
::ArrowArray c_array;
@@ -240,7 +241,8 @@ TEST_F(TantivyGlobalIndexWriterTest,
RejectsHmmTokenizeMode) {
auto root_dir = paimon::test::UniqueTestDirectory::Create();
ASSERT_TRUE(root_dir);
auto path_factory =
std::make_shared<FakeIndexPathFactory>(root_dir->Str());
- auto file_writer = std::make_shared<GlobalIndexFileManager>(fs_,
path_factory);
+ auto file_writer = std::make_shared<GlobalIndexFileManager>(
+ fs_, path_factory, /*checkpoint_path_factory=*/nullptr);
// hmm rejection only fires when the jieba tokenizer is actually
constructed,
// so this test must explicitly opt into jieba (default tokenizer skips
// jieba construction entirely).
diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp
index 355220c7..1a895e49 100644
--- a/test/inte/global_index_test.cpp
+++ b/test/inte/global_index_test.cpp
@@ -159,11 +159,12 @@ class GlobalIndexTest : public ::testing::Test, public
::testing::WithParamInter
const std::string& index_field_name, const std::string&
index_type,
const std::map<std::string, std::string>& options, const
Range& range) {
PAIMON_ASSIGN_OR_RAISE(auto split, ScanData(table_path,
partition_filters));
- PAIMON_ASSIGN_OR_RAISE(auto index_commit_msg,
GlobalIndexWriteTask::WriteIndex(
- table_path,
index_field_name, index_type,
-
std::make_shared<IndexedSplitImpl>(
- split,
std::vector<Range>({range})),
- options, pool_,
fs_));
+ PAIMON_ASSIGN_OR_RAISE(
+ auto index_commit_msg,
+ GlobalIndexWriteTask::WriteIndex(
+ table_path, index_field_name, index_type,
+ std::make_shared<IndexedSplitImpl>(split,
std::vector<Range>({range})), options,
+ /*task_id=*/std::nullopt, pool_, fs_));
return Commit(table_path, {index_commit_msg});
}
@@ -292,7 +293,8 @@ TEST_P(GlobalIndexTest, TestWriteLuminaIndex) {
table_path, "f1", "lumina",
std::make_shared<IndexedSplitImpl>(
split,
std::vector<Range>({Range(0, 3)})),
-
/*options=*/lumina_options, pool_));
+ /*options=*/lumina_options,
+ /*task_id=*/std::nullopt,
pool_));
auto index_commit_msg_impl =
std::dynamic_pointer_cast<CommitMessageImpl>(index_commit_msg);
ASSERT_TRUE(index_commit_msg_impl);
@@ -313,6 +315,67 @@ TEST_P(GlobalIndexTest, TestWriteLuminaIndex) {
ASSERT_TRUE(expected_commit_message->TEST_Equal(*index_commit_msg_impl));
}
+TEST_P(GlobalIndexTest, TestWriteLuminaIndexWithCheckpoint) {
+ arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
+ arrow::field("f1",
arrow::list(arrow::float32()))};
+ auto schema = arrow::schema(fields);
+ std::map<std::string, std::string> lumina_options = {
+ {"lumina.index.dimension", "4"},
+ {"lumina.index.type", "bruteforce"},
+ {"lumina.distance.metric", "l2"},
+ {"lumina.encoding.type", "rawf32"},
+ {"lumina.extension.build.ckpt.count", "1"},
+ {"lumina.extension.build.ckpt.threshold", "1"},
+ {"lumina.search.parallel_number", "10"}};
+
+ std::map<std::string, std::string> options = {{Options::FILE_FORMAT,
file_format_},
+ {Options::FILE_SYSTEM,
"local"},
+
{Options::ROW_TRACKING_ENABLED, "true"},
+
{Options::DATA_EVOLUTION_ENABLED, "true"}};
+
+ CreateTable(/*partition_keys=*/{}, schema, options);
+ std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+ std::vector<std::string> write_cols = schema->field_names();
+ auto src_array =
arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+ ["a", [0.0, 0.0, 0.0, 0.0]],
+ ["b", [0.0, 1.0, 0.0, 1.0]],
+ ["c", [1.0, 0.0, 1.0, 0.0]],
+ ["d", [1.0, 1.0, 1.0, 1.0]]
+
+ ])")
+ .ValueOrDie();
+
+ ASSERT_OK_AND_ASSIGN(auto commit_msgs, WriteArray(table_path, write_cols,
src_array));
+ ASSERT_OK(Commit(table_path, commit_msgs));
+
+ 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");
+ ASSERT_OK(checkpoint_fs->Mkdirs(checkpoint_dir));
+ auto checkpoint_path = [&](const std::string& task, int64_t id) {
+ return PathUtil::JoinPath(
+ checkpoint_dir,
fmt::format("lumina-global-index-f1-0-3-{}-{}.index.ckpt", task, id));
+ };
+ 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(checkpoint_dir, &checkpoint_files));
+ ASSERT_EQ(checkpoint_files.size(), 1);
+ ASSERT_EQ(PathUtil::GetName(checkpoint_files[0].GetPath()),
+ PathUtil::GetName(unrelated_checkpoint));
+}
+
TEST_P(GlobalIndexTest, TestWriteLuminaIndexWithMismatchedDimension) {
arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
arrow::field("f1",
arrow::list(arrow::float32()))};
@@ -438,11 +501,12 @@ TEST_P(GlobalIndexTest, TestWriteIndex) {
ASSERT_OK(Commit(table_path, commit_msgs));
ASSERT_OK_AND_ASSIGN(auto split, ScanData(table_path,
/*partition_filters=*/{}));
- ASSERT_OK_AND_ASSIGN(auto index_commit_msg,
GlobalIndexWriteTask::WriteIndex(
- table_path, "f0", "bitmap",
-
std::make_shared<IndexedSplitImpl>(
- split,
std::vector<Range>({Range(0, 7)})),
- /*options=*/{}, pool_));
+ ASSERT_OK_AND_ASSIGN(
+ auto index_commit_msg,
+ GlobalIndexWriteTask::WriteIndex(
+ table_path, "f0", "bitmap",
+ std::make_shared<IndexedSplitImpl>(split,
std::vector<Range>({Range(0, 7)})),
+ /*options=*/{}, /*task_id=*/std::nullopt, pool_));
auto index_commit_msg_impl =
std::dynamic_pointer_cast<CommitMessageImpl>(index_commit_msg);
ASSERT_TRUE(index_commit_msg_impl);
@@ -466,7 +530,7 @@ TEST_P(GlobalIndexTest, TestWriteIndex) {
GlobalIndexWriteTask::WriteIndex(
table_path, "f0", "invalid",
std::make_shared<IndexedSplitImpl>(split,
std::vector<Range>({Range(0, 7)})),
- /*options=*/{}, pool_),
+ /*options=*/{}, /*task_id=*/std::nullopt, pool_),
"Unknown index type invalid, may not registered");
}
{
@@ -475,7 +539,7 @@ TEST_P(GlobalIndexTest, TestWriteIndex) {
table_path, "f0", "bitmap",
std::make_shared<IndexedSplitImpl>(
split, std::vector<Range>({Range(0, 6),
Range(7, 7)})),
- /*options=*/{}, pool_),
+ /*options=*/{}, /*task_id=*/std::nullopt,
pool_),
"GlobalIndexWriteTask only supports a single
contiguous range.");
}
}
@@ -517,7 +581,7 @@ TEST_P(GlobalIndexTest, TestWriteIndexWithPartition) {
GlobalIndexWriteTask::WriteIndex(
table_path, "f0", "bitmap",
std::make_shared<IndexedSplitImpl>(split,
std::vector<Range>({expected_range})),
- /*options=*/{}, pool_));
+ /*options=*/{}, /*task_id=*/std::nullopt, pool_));
auto index_commit_msg_impl =
std::dynamic_pointer_cast<CommitMessageImpl>(index_commit_msg);
ASSERT_TRUE(index_commit_msg_impl);
@@ -1273,7 +1337,7 @@ TEST_P(GlobalIndexTest,
TestWriteAndQueryLuminaIndexWithTagNullAndEmptyValues) {
GlobalIndexWriteTask::WriteIndex(
table_path, "embedding", "lumina",
std::make_shared<IndexedSplitImpl>(split,
std::vector<Range>({Range(0, 4)})),
- /*options=*/lumina_options, pool_, fs_));
+ /*options=*/lumina_options, /*task_id=*/std::nullopt, pool_, fs_));
std::shared_ptr<CommitMessageImpl> index_commit_msg_impl =
std::dynamic_pointer_cast<CommitMessageImpl>(index_commit_msg);
