From 97e7c426a072f7f85fe3b9521c8de4c69c3f1b0b Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Wed, 23 Sep 2026 10:28:11 +0800 Subject: [PATCH 1/4] feat(global-index): support checkpoints for Lumina index builds --- .../global_index/global_index_write_task.h | 9 +- include/paimon/global_index/global_indexer.h | 5 + .../io/global_index_checkpoint_file_manager.h | 53 +++ src/paimon/CMakeLists.txt | 1 + .../global_indexer_factory_test.cpp | 1 + .../global_index/global_index_file_manager.h | 120 ++++- .../global_index_file_manager_test.cpp | 293 ++++++++++++ .../global_index/global_index_write_task.cpp | 38 +- .../index/index_checkpoint_path_factory.h | 49 ++ .../core/utils/file_store_path_factory.cpp | 82 ++++ .../core/utils/file_store_path_factory.h | 5 + .../utils/file_store_path_factory_test.cpp | 138 ++++++ src/paimon/global_index/lumina/CMakeLists.txt | 5 +- .../lumina/lumina_checkpoint_manager.cpp | 55 +++ .../lumina/lumina_checkpoint_manager.h | 45 ++ .../lumina/lumina_global_index.cpp | 152 ++++-- .../global_index/lumina/lumina_global_index.h | 36 +- .../lumina/lumina_global_index_test.cpp | 437 +++++++++++++++++- test/inte/global_index_test.cpp | 94 +++- 19 files changed, 1533 insertions(+), 85 deletions(-) create mode 100644 include/paimon/global_index/io/global_index_checkpoint_file_manager.h create mode 100644 src/paimon/core/global_index/global_index_file_manager_test.cpp create mode 100644 src/paimon/core/index/index_checkpoint_path_factory.h create mode 100644 src/paimon/global_index/lumina/lumina_checkpoint_manager.cpp create mode 100644 src/paimon/global_index/lumina/lumina_checkpoint_manager.h diff --git a/include/paimon/global_index/global_index_write_task.h b/include/paimon/global_index/global_index_write_task.h index 33760405b..0b2f36a76 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 #include +#include #include #include "paimon/global_index/indexed_split.h" @@ -46,6 +47,11 @@ class PAIMON_EXPORT GlobalIndexWriteTask { /// by the given `indexed_split`. /// @param options Index-specific configuration (e.g., false positive rate for bloom /// 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 +61,8 @@ class PAIMON_EXPORT GlobalIndexWriteTask { static Result> WriteIndex( const std::string& table_path, const std::string& field_name, const std::string& index_type, const std::shared_ptr& indexed_split, - const std::map& options, const std::shared_ptr& pool, + const std::map& options, + const std::optional& task_id, const std::shared_ptr& pool, const std::shared_ptr& file_system = nullptr); }; diff --git a/include/paimon/global_index/global_indexer.h b/include/paimon/global_index/global_indexer.h index 4da6293ff..5be789aca 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& file_reader, const std::vector& files, const std::shared_ptr& 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 000000000..7a07b9aeb --- /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 + +#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> CreateCheckpointOutputStream() const = 0; + + /// Opens the matching checkpoint with the largest numeric id for reading. + virtual Result> OpenCheckpointInputStream() const = 0; + + /// Returns whether the checkpoint file exists. + virtual Result 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 ffd96c70d..6fc99343d 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/global_indexer_factory_test.cpp b/src/paimon/common/global_index/global_indexer_factory_test.cpp index c03fec581..30c58b830 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 indexer, GlobalIndexerFactory::Get("bitmap", options)); ASSERT_TRUE(dynamic_cast(indexer.get())); + ASSERT_FALSE(indexer->SupportsCheckpoint()); } TEST(GlobalIndexerFactoryTest, TestNonExist) { 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 3db9965b2..62ed14206 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 +#include #include +#include #include +#include +#include +#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& fs, - const std::shared_ptr& path_factory) - : fs_(fs), path_factory_(path_factory) {} + GlobalIndexFileManager( + const std::shared_ptr& fs, + const std::shared_ptr& path_factory, + std::unique_ptr checkpoint_path_factory = nullptr) + : fs_(fs), + path_factory_(path_factory), + checkpoint_path_factory_(std::move(checkpoint_path_factory)) {} Result> GetInputStream( const std::string& file_path) const override { @@ -71,8 +86,105 @@ class GlobalIndexFileManager : public GlobalIndexFileReader, public GlobalIndexF return path_factory_->IsExternalPath(); } + bool SupportsCheckpoint() const override { + return checkpoint_path_factory_ != nullptr; + } + + Result> CreateCheckpointOutputStream() const override { + if (!SupportsCheckpoint()) { + return Status::Invalid("global index checkpoint storage is not configured"); + } + PAIMON_ASSIGN_OR_RAISE(std::string file_name, NewCheckpointFileName()); + PAIMON_RETURN_NOT_OK(fs_->Mkdirs(checkpoint_path_factory_->GetDirectoryPath())); + return fs_->Create(checkpoint_path_factory_->ToPath(file_name), /*overwrite=*/false); + } + + Result> OpenCheckpointInputStream() const override { + if (!SupportsCheckpoint()) { + return Status::Invalid("global index checkpoint storage is not configured"); + } + PAIMON_ASSIGN_OR_RAISE(std::optional checkpoint_file, + LatestCheckpointFile()); + if (!checkpoint_file) { + return Status::NotExist("global index checkpoint file does not exist"); + } + return fs_->Open(checkpoint_file->path); + } + + Result CheckpointExists() const override { + if (!SupportsCheckpoint()) { + return Status::Invalid("global index checkpoint storage is not configured"); + } + PAIMON_ASSIGN_OR_RAISE(std::optional 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 checkpoint_files, ListCheckpointFiles()); + for (const CheckpointFile& checkpoint_file : checkpoint_files) { + PAIMON_RETURN_NOT_OK(fs_->Delete(checkpoint_file.path, /*recursive=*/false)); + } + return Status::OK(); + } + private: + struct CheckpointFile { + int64_t id; + std::string path; + }; + + Result NewCheckpointFileName() const { + if (!checkpoint_file_id_initialized_) { + PAIMON_ASSIGN_OR_RAISE(std::optional checkpoint_file, + LatestCheckpointFile()); + checkpoint_path_factory_->InitializeFileId(checkpoint_file ? checkpoint_file->id : -1); + checkpoint_file_id_initialized_ = true; + } + PAIMON_ASSIGN_OR_RAISE(std::string path, checkpoint_path_factory_->NewPath()); + return PathUtil::GetName(path); + } + + Result> ListCheckpointFiles() const { + std::vector file_statuses; + PAIMON_RETURN_NOT_OK( + fs_->ListDir(checkpoint_path_factory_->GetDirectoryPath(), &file_statuses)); + std::vector 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 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> LatestCheckpointFile() const { + PAIMON_ASSIGN_OR_RAISE(std::vector checkpoint_files, ListCheckpointFiles()); + if (checkpoint_files.empty()) { + return std::optional(); + } + 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(std::move(latest)); + } + std::shared_ptr fs_; std::shared_ptr path_factory_; + std::unique_ptr checkpoint_path_factory_; + // Historical ids are loaded only on the first successful file name allocation scan. + mutable bool checkpoint_file_id_initialized_ = false; }; } // 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 000000000..e1a008128 --- /dev/null +++ b/src/paimon/core/global_index/global_index_file_manager_test.cpp @@ -0,0 +1,293 @@ +/* + * 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 +#include +#include + +#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* 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> CreateManager() const { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr checkpoint_path_factory, + path_factory_->CreateGlobalIndexCheckpointPathFactory( + "lumina", "vector", Range(10, 20), "task-1")); + return std::make_shared( + fs_, path_factory_->CreateGlobalIndexFileFactory(), std::move(checkpoint_path_factory)); + } + + Result CreateCheckpointFile( + const std::shared_ptr& manager) const { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr 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 fs_ = std::make_shared(); + std::unique_ptr dir_; + std::shared_ptr path_factory_; +}; + +TEST_F(GlobalIndexFileManagerTest, TestIndexIOWithoutCheckpointAccess) { + fs_->fail_list_ = true; + ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); + std::shared_ptr writer = manager; + std::shared_ptr reader = manager; + ASSERT_TRUE(std::dynamic_pointer_cast(writer)); + ASSERT_TRUE(std::dynamic_pointer_cast(reader)); + ASSERT_TRUE(manager->SupportsCheckpoint()); + auto plain_manager = std::make_shared( + fs_, path_factory_->CreateGlobalIndexFileFactory()); + ASSERT_TRUE(std::dynamic_pointer_cast(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 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 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()); + 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 manager, CreateManager()); + ASSERT_EQ(fs_->list_count_, 0); + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/10_20"); + 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 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 restarted, CreateManager()); + ASSERT_OK_AND_ASSIGN(bool exists, restarted->CheckpointExists()); + ASSERT_TRUE(exists); + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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 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 manager, CreateManager()); + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/10_20"); + 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 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 manager, CreateManager()); + ASSERT_OK_AND_ASSIGN(bool exists, manager->CheckpointExists()); + ASSERT_FALSE(exists); + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/10_20"); + 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 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 manager, CreateManager()); + std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/10_20"); + std::string prefix = "lumina-global-index-vector-10-20-task-1-"; + int64_t max_id = std::numeric_limits::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 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 input, + restarted->OpenCheckpointInputStream()); + ASSERT_OK(input->Close()); + ASSERT_OK(restarted->DeleteCheckpoint()); +} + +TEST_F(GlobalIndexFileManagerTest, TestSerialAllocations) { + ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); + std::vector 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(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 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_write_task.cpp b/src/paimon/core/global_index/global_index_write_task.cpp index 0288d1b96..59ca152fb 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> CreateGlobalIndexer(const std::string& in return indexer; } -Result> CreateGlobalIndexFileManager( +Result> CreateFileStorePathFactory( const std::string& table_path, const std::shared_ptr& table_schema, const CoreOptions& core_options, const std::shared_ptr& pool) { auto all_arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields()); @@ -65,18 +66,11 @@ Result> CreateGlobalIndexFileManager( core_options.CreateExternalPaths()); PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, core_options.CreateGlobalIndexExternalPath()); - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr 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 index_path_factory = - path_factory->CreateGlobalIndexFileFactory(); - return std::make_shared(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> CreateGlobalIndexWriter( @@ -344,7 +338,7 @@ Result> ToCommitMessage( Result> GlobalIndexWriteTask::WriteIndex( const std::string& table_path, const std::string& field_name, const std::string& index_type, const std::shared_ptr& indexed_split, - const std::map& options, + const std::map& options, const std::optional& task_id, const std::shared_ptr& memory_pool, const std::shared_ptr& file_system) { auto data_split = std::dynamic_pointer_cast(indexed_split->GetDataSplit()); @@ -387,12 +381,20 @@ Result> GlobalIndexWriteTask::WriteIndex( std::vector writer_field_names = BuildWriterFieldNames(field_name, extra_fields); std::vector 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 index_file_manager, - CreateGlobalIndexFileManager(table_path, table_schema, core_options, pool)); + std::shared_ptr path_factory, + CreateFileStorePathFactory(table_path, table_schema, core_options, pool)); + std::unique_ptr 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( + core_options.GetFileSystem(), path_factory->CreateGlobalIndexFileFactory(), + std::move(checkpoint_path_factory)); - // create batch reader PAIMON_ASSIGN_OR_RAISE( std::unique_ptr batch_reader, CreateBatchReader(table_path, read_field_names, indexed_split, core_options, pool)); 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 000000000..e5d2d59b6 --- /dev/null +++ b/src/paimon/core/index/index_checkpoint_path_factory.h @@ -0,0 +1,49 @@ +/* + * 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 +#include +#include + +#include "paimon/result.h" + +namespace paimon { + +/// Path factory for global index checkpoints scoped to one task. +class IndexCheckpointPathFactory { + public: + virtual ~IndexCheckpointPathFactory() = default; + + /// Initializes the counter before the first allocation, without storage I/O. + /// @param last_file_id Largest existing checkpoint id, or -1 if no checkpoint exists. + virtual void InitializeFileId(int64_t last_file_id) = 0; + + /// Allocates the next id and creates its path, without storage I/O. Fails on id overflow. + virtual Result NewPath() 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 GetCheckpointId(const std::string& file_name) const = 0; +}; + +} // namespace paimon diff --git a/src/paimon/core/utils/file_store_path_factory.cpp b/src/paimon/core/utils/file_store_path_factory.cpp index ef4aae42d..b38f530cf 100644 --- a/src/paimon/core/utils/file_store_path_factory.cpp +++ b/src/paimon/core/utils/file_store_path_factory.cpp @@ -19,9 +19,13 @@ #include "paimon/core/utils/file_store_path_factory.h" #include +#include +#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 +35,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 +44,31 @@ class Schema; namespace paimon { class MemoryPool; +namespace { + +constexpr char kIndexCheckpointFileSuffix[] = ".index.ckpt"; + +std::optional 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::length(kIndexCheckpointFileSuffix); + if (suffix_pos <= file_name_prefix.size()) { + return std::nullopt; + } + std::optional id = StringUtils::StringToValue( + 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 +213,58 @@ std::unique_ptr FileStorePathFactory::CreateGlobalIndexFileFac return std::make_unique(shared_from_this()); } +Result> +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) {} + + void InitializeFileId(int64_t last_file_id) override { + assert(last_file_id >= -1); + last_file_id_ = last_file_id; + } + + Result NewPath() const override { + if (last_file_id_ == std::numeric_limits::max()) { + return Status::Invalid("checkpoint file id exceeds int64 max"); + } + std::string file_name = fmt::format("{}{}{}", file_name_prefix_, ++last_file_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 GetCheckpointId(const std::string& file_name) const override { + return ParseCheckpointId(file_name, file_name_prefix_); + } + + private: + std::string directory_; + std::string file_name_prefix_; + mutable int64_t last_file_id_ = -1; + }; + 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( + std::make_unique(directory, file_name_prefix)); +} + Result> FileStorePathFactory::CreateDataFilePathFactory( const BinaryRow& partition, int32_t bucket) const { auto data_file_path_factory = std::make_shared(); diff --git a/src/paimon/core/utils/file_store_path_factory.h b/src/paimon/core/utils/file_store_path_factory.h index 1890f2bc3..030b6b78c 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 { public: @@ -76,6 +78,9 @@ class FileStorePathFactory : public std::enable_shared_from_this> CreateIndexFileFactory(const BinaryRow& partition, int32_t bucket); std::unique_ptr CreateGlobalIndexFileFactory(); + Result> CreateGlobalIndexCheckpointPathFactory( + const std::string& index_type, const std::string& field_name, const Range& range, + const std::string& task_id); Result> CreateDataFilePathFactory( const BinaryRow& partition, int32_t bucket) const; Result ToBinaryRow(const std::map& 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 c78db99df..67bf039c3 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 +#include #include #include #include @@ -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,135 @@ 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 file_store_path_factory = CreateFactory(dir->Str()); + std::shared_ptr file_system = dir->GetFileSystem(); + + std::string prefix = "lumina-global-index-vector-10-20-task-1-"; + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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); + checkpoint_path_factory->InitializeFileId(-1); + ASSERT_OK_AND_ASSIGN(std::string first_path, checkpoint_path_factory->NewPath()); + ASSERT_EQ(first_path, PathUtil::JoinPath(checkpoint_dir, prefix + "0.index.ckpt")); + ASSERT_OK_AND_ASSIGN(std::string second_path, checkpoint_path_factory->NewPath()); + 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, TestCheckpointFileIdInitializationAndOverflow) { + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + auto factory = CreateFactory(dir->Str()); + for (int64_t last_id : {int64_t{9}, std::numeric_limits::max() - 1, + std::numeric_limits::max()}) { + ASSERT_OK_AND_ASSIGN(std::unique_ptr checkpoint_factory, + factory->CreateGlobalIndexCheckpointPathFactory( + "lumina", "vector", Range(10, 20), "task-1")); + checkpoint_factory->InitializeFileId(last_id); + if (last_id != std::numeric_limits::max()) { + ASSERT_OK_AND_ASSIGN(std::string path, checkpoint_factory->NewPath()); + ASSERT_EQ(checkpoint_factory->GetCheckpointId(PathUtil::GetName(path)), last_id + 1); + } + if (last_id >= std::numeric_limits::max() - 1) { + ASSERT_NOK_WITH_MSG(checkpoint_factory->NewPath(), + "checkpoint file id exceeds int64 max"); + ASSERT_NOK_WITH_MSG(checkpoint_factory->NewPath(), + "checkpoint file id exceeds int64 max"); + } + 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 fs = dir->GetFileSystem(); + std::shared_ptr factory = CreateFactory(dir->Str()); + ASSERT_OK_AND_ASSIGN(std::unique_ptr own_factory, + factory->CreateGlobalIndexCheckpointPathFactory("lumina", "vector", + Range(10, 20), "task-1")); + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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())); + ASSERT_OK_AND_ASSIGN(std::string own_path, own_factory->NewPath()); + ASSERT_OK(fs->WriteFile(own_path, "checkpoint", /*overwrite=*/false)); + other_factory->InitializeFileId(99); + ASSERT_OK_AND_ASSIGN(std::string other_path, other_factory->NewPath()); + 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 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 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{"", ".", "..", " ", "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/lumina/CMakeLists.txt b/src/paimon/global_index/lumina/CMakeLists.txt index b0496df65..ccf6a5e5b 100644 --- a/src/paimon/global_index/lumina/CMakeLists.txt +++ b/src/paimon/global_index/lumina/CMakeLists.txt @@ -15,7 +15,10 @@ # 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 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 000000000..93187edac --- /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 + +#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> output = file_manager_->CreateCheckpointOutputStream(); + if (!output.ok()) { + return nullptr; + } + std::shared_ptr shared_output = std::move(output).value(); + return std::make_unique(shared_output); +} + +::lumina::core::Result LuminaCheckpointManager::HasCkptFile() { + Result exists = file_manager_->CheckpointExists(); + if (!exists.ok()) { + return ::lumina::core::Result::Err(PaimonToLuminaStatus(exists.status())); + } + return ::lumina::core::Result::Ok(exists.value()); +} + +std::unique_ptr<::lumina::io::FileReader> LuminaCheckpointManager::GetCkptFileReader() { + Result> input = file_manager_->OpenCheckpointInputStream(); + if (!input.ok()) { + return nullptr; + } + std::shared_ptr shared_input = std::move(input).value(); + return std::make_unique(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 000000000..e52316e21 --- /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 + +#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& file_manager) + : file_manager_(file_manager) {} + + std::unique_ptr<::lumina::io::FileWriter> CreateCkptFileWriter() override; + + ::lumina::core::Result HasCkptFile() override; + + std::unique_ptr<::lumina::io::FileReader> GetCkptFileReader() override; + + private: + std::shared_ptr 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 98585f21d..16863f128 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 #include #include +#include #include #include #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 LiteralsToTagValues(const std::vector& literals) { } } +bool IsCheckpointEnabled(const std::map& lumina_options) { + return lumina_options.count(std::string(::lumina::core::kExtensionCkptThreshold)) != 0 || + lumina_options.count(std::string(::lumina::core::kExtensionCkptCount)) != 0; +} + } // namespace Result> LuminaIndexWriter::ExtractTagDataForSegment( @@ -582,10 +587,19 @@ Result> LuminaGlobalIndex::CreateWriter( ::lumina::api::BuilderOptions builder_options, ::lumina::api::NormalizeBuilderOptions(std::unordered_map( lumina_options.begin(), lumina_options.end()))); + std::shared_ptr checkpoint_file_manager; + if (IsCheckpointEnabled(lumina_options)) { + checkpoint_file_manager = + std::dynamic_pointer_cast(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(pool); return std::make_shared( 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::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_type, uint32_t dimension, const std::shared_ptr& file_manager, ::lumina::api::BuilderOptions&& builder_options, ::lumina::api::IOOptions&& io_options, const std::map& lumina_options, - std::vector&& tag_fields, const std::shared_ptr& pool) + std::vector&& tag_fields, + const std::shared_ptr& checkpoint_file_manager, + const std::shared_ptr& 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&& 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,95 @@ Result> LuminaIndexWriter::Finish() { if (indexed_count_ == 0) { return std::vector(); } - ::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>().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>().swap(array_vec_); - std::vector>().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> { + ::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(std::move(builder)); + if (checkpoint_file_manager_) { + auto checkpoint_manager = + std::make_unique(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(); + }; + Status checkpoint_status = Status::OK(); + if (tag_fields_.empty()) { + context->checkpoint_extension = std::make_unique< + ::lumina::extensions::experimental::BuildWithCheckpointExtension>(); + checkpoint_status = attach_checkpoint(context->checkpoint_extension.get()); + } else { + context->checkpoint_tag_extension = std::make_unique< + ::lumina::extensions::experimental::BuildWithCkptAndTagExtension>(); + checkpoint_status = attach_checkpoint(context->checkpoint_tag_extension.get()); + } + if (!checkpoint_status.ok()) { + return checkpoint_status; + } + } 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> { + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr 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> 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 build_context = std::move(build_result).value(); + std::vector>().swap(array_vec_); + std::vector>().swap(tag_data_vec_); + // dump index PAIMON_ASSIGN_OR_RAISE(std::string index_file_name, file_manager_->NewFileName(LuminaDefines::kIdentifier)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr out, file_manager_->NewOutputStream(index_file_name)); auto file_writer = std::make_unique(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 +1019,14 @@ Result> LuminaIndexWriter::Finish() { auto meta_bytes = std::make_shared(options_json, pool_->GetPaimonPool().get()); GlobalIndexIOMeta meta(file_manager_->ToPath(index_file_name), file_size, /*metadata=*/meta_bytes); + if (checkpoint_file_manager_) { + PAIMON_RETURN_NOT_OK(checkpoint_file_manager_->DeleteCheckpoint()); + } return std::vector({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& pool) diff --git a/src/paimon/global_index/lumina/lumina_global_index.h b/src/paimon/global_index/lumina/lumina_global_index.h index c2c30475a..b47873743 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 #include #include #include #include -#include -#include #include #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& options) : options_(options) {} + bool SupportsCheckpoint() const override { + return true; + } + Result>> GetExtraFieldNames() const override; + /// With checkpoints enabled, file_writer must implement GlobalIndexCheckpointFileManager and + /// return true from SupportsCheckpoint(). Result> CreateWriter( const std::string& field_name, ::ArrowSchema* arrow_schema, const std::shared_ptr& 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_type, uint32_t dimension, - const std::shared_ptr& file_manager, - ::lumina::api::BuilderOptions&& builder_options, - ::lumina::api::IOOptions&& io_options, - const std::map& lumina_options, - std::vector&& tag_fields, - const std::shared_ptr& pool); + LuminaIndexWriter( + const std::string& field_name, const std::shared_ptr& arrow_type, + uint32_t dimension, const std::shared_ptr& file_manager, + ::lumina::api::BuilderOptions&& builder_options, ::lumina::api::IOOptions&& io_options, + const std::map& lumina_options, + std::vector&& tag_fields, + const std::shared_ptr& checkpoint_file_manager, + const std::shared_ptr& pool); Status AddBatch(::ArrowArray* arrow_array, std::vector&& relative_row_ids) override; @@ -141,6 +152,7 @@ class LuminaIndexWriter : public GlobalIndexWriter { ::lumina::api::IOOptions io_options_; std::map lumina_options_; std::vector tag_fields_; + std::shared_ptr checkpoint_file_manager_; std::vector> array_vec_; std::vector array_start_ids_; std::vector> tag_data_vec_; @@ -249,7 +261,7 @@ class LuminaIndexReader : public GlobalIndexReader { static Result<::lumina::extensions::experimental::TagFilter> PredicateToTagFilter( const std::shared_ptr& predicate); - LuminaIndexReader::IndexInfo index_info_; + IndexInfo index_info_; std::shared_ptr 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 2d54950ad..c80ecaf65 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 CheckpointExists() const override { + check_checkpoint_count_++; + return GlobalIndexFileManager::CheckpointExists(); + } + + Result> CreateCheckpointOutputStream() const override { + create_checkpoint_count_++; + return GlobalIndexFileManager::CreateCheckpointOutputStream(); + } + + Result> 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& 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 GetNextBatch( + std::vector& vector_buffer, + std::vector<::lumina::core::vector_id_t>& id_buffer) noexcept override { + if (consumed_) { + return ::lumina::core::Result::Ok(0); + } + vector_buffer = vectors_; + id_buffer = ids_; + consumed_ = true; + return ::lumina::core::Result::Ok(ids_.size()); + } + + private: + uint32_t dimension_; + std::vector vectors_; + std::vector<::lumina::core::vector_id_t> ids_; + bool consumed_ = false; + }; + std::unique_ptr<::ArrowSchema> CreateArrowSchema( const std::shared_ptr& data_type) const { auto c_schema = std::make_unique<::ArrowSchema>(); @@ -69,14 +134,47 @@ class LuminaGlobalIndexTest : public ::testing::Test { return c_schema; } + Result> 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> 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 file_store_path_factory, + CreateFileStorePathFactory(table_path)); + return file_store_path_factory->CreateGlobalIndexCheckpointPathFactory( + index_type, field_name, range, "task-1"); + } + + Result> CreateCountingCheckpointFileManager( + const std::shared_ptr& path_factory, const Range& range) const { + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr checkpoint_path_factory, + path_factory->CreateGlobalIndexCheckpointPathFactory("lumina", "f0", range, "task-1")); + return std::make_shared( + fs_, path_factory->CreateGlobalIndexFileFactory(), std::move(checkpoint_path_factory)); + } + Result WriteGlobalIndex(const std::string& index_root, const std::shared_ptr& data_type, const std::map& options, const std::shared_ptr& array, const Range& expected_range) const { auto global_index = std::make_shared(options); - auto path_factory = std::make_shared(index_root); - auto file_writer = std::make_shared(fs_, path_factory); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr path_factory, + CreateFileStorePathFactory(index_root)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr checkpoint_path_factory, + path_factory->CreateGlobalIndexCheckpointPathFactory( + "lumina", "f0", expected_range, "task-1")); + auto file_writer = std::make_shared( + fs_, path_factory->CreateGlobalIndexFileFactory(), std::move(checkpoint_path_factory)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr global_writer, global_index->CreateWriter("f0", CreateArrowSchema(data_type).get(), @@ -252,12 +350,343 @@ 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 path_factory, + CreateFileStorePathFactory(dir->Str())); + auto plain_manager = + std::make_shared(fs_, path_factory->CreateGlobalIndexFileFactory()); + ASSERT_OK_AND_ASSIGN(std::shared_ptr checkpoint_manager, + CreateCountingCheckpointFileManager(path_factory, Range(0, 3))); + std::map 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{"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 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(checkpoint_options); + ASSERT_OK_AND_ASSIGN(std::shared_ptr path_factory, + CreateFileStorePathFactory(test_root)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr file_manager, + CreateCountingCheckpointFileManager(path_factory, Range(0, 3))); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr 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 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 reader, + CreateGlobalIndexReader(test_root, data_type_, checkpoint_options, result_metas[0])); + ASSERT_OK_AND_ASSIGN(std::shared_ptr scored_result, + reader->VisitVectorSearch(std::make_shared( + /*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 checkpoint_path_factory, + CreateCheckpointPathFactory(test_root, "lumina", "f0", Range(10, 20))); + ASSERT_OK_AND_ASSIGN(std::shared_ptr path_factory, + CreateFileStorePathFactory(test_root)); + auto file_manager = std::make_shared( + 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 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 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( + fs_, path_factory->CreateGlobalIndexFileFactory(), std::move(checkpoint_path_factory)); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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 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 path_factory, + CreateFileStorePathFactory(test_root)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr file_manager, + CreateCountingCheckpointFileManager(path_factory, Range(0, 3))); + + std::map checkpoint_options = options_; + checkpoint_options["lumina.extension.build.ckpt.threshold"] = "1"; + checkpoint_options["lumina.extension.build.ckpt.count"] = "1"; + std::vector 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(4)) + .Set(::lumina::core::kDistanceMetric, ::lumina::core::kDistanceL2) + .Set(::lumina::core::kEncodingType, ::lumina::core::kEncodingRawf32) + .Set(::lumina::core::kExtensionCkptThreshold, static_cast(1)) + .Set(::lumina::core::kExtensionCkptCount, static_cast(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(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(checkpoint_options); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr 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 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 reader, + CreateGlobalIndexReader(test_root, data_type_, checkpoint_options, result_metas[0])); + ASSERT_OK_AND_ASSIGN(std::shared_ptr scored_result, + reader->VisitVectorSearch(std::make_shared( + /*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 path_factory, + CreateFileStorePathFactory(test_root)); + ASSERT_OK_AND_ASSIGN(std::shared_ptr file_manager, + CreateCountingCheckpointFileManager(path_factory, Range(0, 3))); + + ASSERT_OK_AND_ASSIGN(std::unique_ptr 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 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(checkpoint_options); + ASSERT_OK_AND_ASSIGN( + std::shared_ptr 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 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 reader, + CreateGlobalIndexReader(test_root, data_type_, checkpoint_options, result_metas[0])); + ASSERT_OK_AND_ASSIGN(std::shared_ptr scored_result, + reader->VisitVectorSearch(std::make_shared( + /*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 path_factory, + CreateFileStorePathFactory(dir->Str())); + ASSERT_OK_AND_ASSIGN(std::unique_ptr checkpoint_path_factory, + path_factory->CreateGlobalIndexCheckpointPathFactory( + "lumina", "f0", Range(0, 3), "task-1")); + auto file_manager = std::make_shared( + 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(4)) + .Set(::lumina::core::kDistanceMetric, ::lumina::core::kDistanceL2) + .Set(::lumina::core::kEncodingType, ::lumina::core::kEncodingRawf32) + .Set(::lumina::core::kExtensionCkptThreshold, static_cast(1)) + .Set(::lumina::core::kExtensionCkptCount, static_cast(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(file_manager)) + .IsOk()); + std::vector 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 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 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 tag_options = options_; + tag_options["lumina.extension.build.ckpt.threshold"] = "1"; + tag_options["lumina.extension.build.ckpt.count"] = "2"; tag_options["lumina.extension.build.tag.tag_schema"] = R"({"key_name":"color","type":"enum","value_type":"string"})"; @@ -276,6 +705,10 @@ 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 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 reader, CreateGlobalIndexReader(test_root, data_type_, tag_options, meta)); diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp index 355220c78..1a895e492 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& 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( - split, std::vector({range})), - options, pool_, fs_)); + PAIMON_ASSIGN_OR_RAISE( + auto index_commit_msg, + GlobalIndexWriteTask::WriteIndex( + table_path, index_field_name, index_type, + std::make_shared(split, std::vector({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( split, std::vector({Range(0, 3)})), - /*options=*/lumina_options, pool_)); + /*options=*/lumina_options, + /*task_id=*/std::nullopt, pool_)); auto index_commit_msg_impl = std::dynamic_pointer_cast(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 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 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 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 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(split, std::vector{Range(0, 3)}), lumina_options, + task_id, pool_)); + std::vector 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( - split, std::vector({Range(0, 7)})), - /*options=*/{}, pool_)); + ASSERT_OK_AND_ASSIGN( + auto index_commit_msg, + GlobalIndexWriteTask::WriteIndex( + table_path, "f0", "bitmap", + std::make_shared(split, std::vector({Range(0, 7)})), + /*options=*/{}, /*task_id=*/std::nullopt, pool_)); auto index_commit_msg_impl = std::dynamic_pointer_cast(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(split, std::vector({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( split, std::vector({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(split, std::vector({expected_range})), - /*options=*/{}, pool_)); + /*options=*/{}, /*task_id=*/std::nullopt, pool_)); auto index_commit_msg_impl = std::dynamic_pointer_cast(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(split, std::vector({Range(0, 4)})), - /*options=*/lumina_options, pool_, fs_)); + /*options=*/lumina_options, /*task_id=*/std::nullopt, pool_, fs_)); std::shared_ptr index_commit_msg_impl = std::dynamic_pointer_cast(index_commit_msg); From d572d1d70057414175515db529b1e336c0378454 Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Wed, 23 Sep 2026 11:06:36 +0800 Subject: [PATCH 2/4] fix ci --- .../core/global_index/global_index_file_manager_test.cpp | 8 ++++---- src/paimon/global_index/lumina/CMakeLists.txt | 7 ++----- .../global_index/lumina/lumina_global_index_test.cpp | 5 ++--- 3 files changed, 8 insertions(+), 12 deletions(-) diff --git a/src/paimon/core/global_index/global_index_file_manager_test.cpp b/src/paimon/core/global_index/global_index_file_manager_test.cpp index e1a008128..646a6b4e2 100644 --- a/src/paimon/core/global_index/global_index_file_manager_test.cpp +++ b/src/paimon/core/global_index/global_index_file_manager_test.cpp @@ -139,7 +139,7 @@ TEST_F(GlobalIndexFileManagerTest, TestCheckpointNotConfigured) { TEST_F(GlobalIndexFileManagerTest, TestCheckpointIOAndRestart) { ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); ASSERT_EQ(fs_->list_count_, 0); - std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/10_20"); + 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. @@ -204,7 +204,7 @@ TEST_F(GlobalIndexFileManagerTest, TestInitializationFailureCanRetry) { TEST_F(GlobalIndexFileManagerTest, TestLatestCheckpointUsesNumericId) { ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); - std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/10_20"); + 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)); @@ -222,7 +222,7 @@ TEST_F(GlobalIndexFileManagerTest, TestReadsDoNotInitializeFileId) { ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); ASSERT_OK_AND_ASSIGN(bool exists, manager->CheckpointExists()); ASSERT_FALSE(exists); - std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/10_20"); + 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)); @@ -242,7 +242,7 @@ TEST_F(GlobalIndexFileManagerTest, TestReadsDoNotInitializeFileId) { TEST_F(GlobalIndexFileManagerTest, TestFileIdOverflow) { ASSERT_OK_AND_ASSIGN(std::shared_ptr manager, CreateManager()); - std::string directory = PathUtil::JoinPath(dir_->Str(), "index/checkpoint/10_20"); + 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::max(); ASSERT_OK(fs_->WriteFile( diff --git a/src/paimon/global_index/lumina/CMakeLists.txt b/src/paimon/global_index/lumina/CMakeLists.txt index ccf6a5e5b..239c77028 100644 --- a/src/paimon/global_index/lumina/CMakeLists.txt +++ b/src/paimon/global_index/lumina/CMakeLists.txt @@ -15,11 +15,8 @@ # limitations under the License. if(PAIMON_ENABLE_LUMINA) - set(PAIMON_LUMINA_INDEX - lumina_checkpoint_manager.cpp - 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_global_index_test.cpp b/src/paimon/global_index/lumina/lumina_global_index_test.cpp index c80ecaf65..e7dc12f26 100644 --- a/src/paimon/global_index/lumina/lumina_global_index_test.cpp +++ b/src/paimon/global_index/lumina/lumina_global_index_test.cpp @@ -686,7 +686,7 @@ TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithTagFilter) { std::map tag_options = options_; tag_options["lumina.extension.build.ckpt.threshold"] = "1"; - tag_options["lumina.extension.build.ckpt.count"] = "2"; + 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"})"; @@ -706,8 +706,7 @@ TEST_F(LuminaGlobalIndexTest, TestWriteAndReadWithTagFilter) { GlobalIndexIOMeta meta, WriteGlobalIndex(test_root, tag_data_type, tag_options, tag_array, Range(0, 3))); std::vector checkpoint_files; - ASSERT_OK( - fs_->ListDir(PathUtil::JoinPath(test_root, "index/checkpoint"), &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 reader, CreateGlobalIndexReader(test_root, data_type_, tag_options, meta)); From 5e1b4616375bbdfd2b1b54ed1843c29ecb5906fc Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Wed, 23 Sep 2026 19:01:06 +0800 Subject: [PATCH 3/4] fix comment --- .../global_index/global_index_write_task.h | 11 +++--- .../bitmap/bitmap_global_index_test.cpp | 6 ++- .../range_bitmap_global_index_test.cpp | 6 ++- .../global_index/global_index_file_manager.h | 37 +++++++++++-------- .../global_index_file_manager_test.cpp | 5 ++- .../global_index/global_index_scan_impl.cpp | 4 +- .../global_index/global_index_write_task.cpp | 1 + .../index/index_checkpoint_path_factory.h | 10 +---- .../core/utils/file_store_path_factory.cpp | 17 ++------- .../utils/file_store_path_factory_test.cpp | 27 ++++---------- .../lucene/lucene_global_index_test.cpp | 6 ++- .../lumina/lumina_global_index.cpp | 16 ++++---- .../lumina/lumina_global_index_test.cpp | 15 +++++--- .../tantivy/tantivy_equivalence_test.cpp | 6 ++- .../tantivy/tantivy_filter_limit_test.cpp | 3 +- .../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 ++- .../tantivy/tantivy_writer_test.cpp | 6 ++- 20 files changed, 102 insertions(+), 98 deletions(-) diff --git a/include/paimon/global_index/global_index_write_task.h b/include/paimon/global_index/global_index_write_task.h index 0b2f36a76..391c02caf 100644 --- a/include/paimon/global_index/global_index_write_task.h +++ b/include/paimon/global_index/global_index_write_task.h @@ -46,12 +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. + /// 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. 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 80e023f97..6e1ad44a9 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(file_index); auto path_factory = std::make_shared(index_root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr global_writer, @@ -112,7 +113,8 @@ class BitmapGlobalIndexTest : public ::testing::Test { auto global_index = std::make_shared(file_index); auto path_factory = std::make_shared(index_root); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + 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/rangebitmap/range_bitmap_global_index_test.cpp b/src/paimon/common/global_index/rangebitmap/range_bitmap_global_index_test.cpp index d19622ad2..c198fbf70 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(file_index); auto path_factory = std::make_shared(index_root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr global_writer, @@ -109,7 +110,8 @@ class RangeBitmapGlobalIndexTest : public ::testing::Test { auto global_index = std::make_shared(file_index); auto path_factory = std::make_shared(index_root); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + 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 62ed14206..8ca5ebf70 100644 --- a/src/paimon/core/global_index/global_index_file_manager.h +++ b/src/paimon/core/global_index/global_index_file_manager.h @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -43,10 +44,9 @@ class GlobalIndexFileManager : public GlobalIndexFileReader, public GlobalIndexFileWriter, public GlobalIndexCheckpointFileManager { public: - GlobalIndexFileManager( - const std::shared_ptr& fs, - const std::shared_ptr& path_factory, - std::unique_ptr checkpoint_path_factory = nullptr) + GlobalIndexFileManager(const std::shared_ptr& fs, + const std::shared_ptr& path_factory, + std::unique_ptr checkpoint_path_factory) : fs_(fs), path_factory_(path_factory), checkpoint_path_factory_(std::move(checkpoint_path_factory)) {} @@ -94,9 +94,9 @@ class GlobalIndexFileManager : public GlobalIndexFileReader, if (!SupportsCheckpoint()) { return Status::Invalid("global index checkpoint storage is not configured"); } - PAIMON_ASSIGN_OR_RAISE(std::string file_name, NewCheckpointFileName()); + 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_->ToPath(file_name), /*overwrite=*/false); + return fs_->Create(checkpoint_path_factory_->NewPath(file_id), /*overwrite=*/false); } Result> OpenCheckpointInputStream() const override { @@ -125,10 +125,14 @@ class GlobalIndexFileManager : public GlobalIndexFileReader, return Status::Invalid("global index checkpoint storage is not configured"); } PAIMON_ASSIGN_OR_RAISE(std::vector checkpoint_files, ListCheckpointFiles()); + Status first_error = Status::OK(); for (const CheckpointFile& checkpoint_file : checkpoint_files) { - PAIMON_RETURN_NOT_OK(fs_->Delete(checkpoint_file.path, /*recursive=*/false)); + Status status = fs_->Delete(checkpoint_file.path, /*recursive=*/false); + if (!status.ok() && first_error.ok()) { + first_error = std::move(status); + } } - return Status::OK(); + return first_error; } private: @@ -137,15 +141,16 @@ class GlobalIndexFileManager : public GlobalIndexFileReader, std::string path; }; - Result NewCheckpointFileName() const { - if (!checkpoint_file_id_initialized_) { + Result NextCheckpointFileId() const { + if (!last_checkpoint_file_id_) { PAIMON_ASSIGN_OR_RAISE(std::optional checkpoint_file, LatestCheckpointFile()); - checkpoint_path_factory_->InitializeFileId(checkpoint_file ? checkpoint_file->id : -1); - checkpoint_file_id_initialized_ = true; + last_checkpoint_file_id_ = checkpoint_file ? checkpoint_file->id : -1; + } + if (last_checkpoint_file_id_.value() == std::numeric_limits::max()) { + return Status::Invalid("checkpoint file id exceeds int64 max"); } - PAIMON_ASSIGN_OR_RAISE(std::string path, checkpoint_path_factory_->NewPath()); - return PathUtil::GetName(path); + return ++last_checkpoint_file_id_.value(); } Result> ListCheckpointFiles() const { @@ -184,7 +189,7 @@ class GlobalIndexFileManager : public GlobalIndexFileReader, std::shared_ptr fs_; std::shared_ptr path_factory_; std::unique_ptr checkpoint_path_factory_; - // Historical ids are loaded only on the first successful file name allocation scan. - mutable bool checkpoint_file_id_initialized_ = false; + // Historical ids are loaded only on the first successful file id allocation scan. + mutable std::optional 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 index 646a6b4e2..ab27ad2e0 100644 --- a/src/paimon/core/global_index/global_index_file_manager_test.cpp +++ b/src/paimon/core/global_index/global_index_file_manager_test.cpp @@ -94,7 +94,7 @@ TEST_F(GlobalIndexFileManagerTest, TestIndexIOWithoutCheckpointAccess) { ASSERT_TRUE(std::dynamic_pointer_cast(reader)); ASSERT_TRUE(manager->SupportsCheckpoint()); auto plain_manager = std::make_shared( - fs_, path_factory_->CreateGlobalIndexFileFactory()); + fs_, path_factory_->CreateGlobalIndexFileFactory(), /*checkpoint_path_factory=*/nullptr); ASSERT_TRUE(std::dynamic_pointer_cast(plain_manager)); ASSERT_FALSE(plain_manager->SupportsCheckpoint()); @@ -122,7 +122,8 @@ TEST_F(GlobalIndexFileManagerTest, TestIndexIOWithoutCheckpointAccess) { TEST_F(GlobalIndexFileManagerTest, TestCheckpointNotConfigured) { fs_->fail_list_ = true; - GlobalIndexFileManager manager(fs_, path_factory_->CreateGlobalIndexFileFactory()); + 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"); 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 e4b3e7b75..c38c50371 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& tab : pool_(pool), table_schema_(table_schema), options_(options), - index_file_manager_( - std::make_shared(options.GetFileSystem(), path_factory)), + index_file_manager_(std::make_shared( + 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 59ca152fb..82c861432 100644 --- a/src/paimon/core/global_index/global_index_write_task.cpp +++ b/src/paimon/core/global_index/global_index_write_task.cpp @@ -395,6 +395,7 @@ Result> GlobalIndexWriteTask::WriteIndex( core_options.GetFileSystem(), path_factory->CreateGlobalIndexFileFactory(), std::move(checkpoint_path_factory)); + // create batch reader PAIMON_ASSIGN_OR_RAISE( std::unique_ptr batch_reader, CreateBatchReader(table_path, read_field_names, indexed_split, core_options, pool)); diff --git a/src/paimon/core/index/index_checkpoint_path_factory.h b/src/paimon/core/index/index_checkpoint_path_factory.h index e5d2d59b6..0c1b13e51 100644 --- a/src/paimon/core/index/index_checkpoint_path_factory.h +++ b/src/paimon/core/index/index_checkpoint_path_factory.h @@ -22,8 +22,6 @@ #include #include -#include "paimon/result.h" - namespace paimon { /// Path factory for global index checkpoints scoped to one task. @@ -31,12 +29,8 @@ class IndexCheckpointPathFactory { public: virtual ~IndexCheckpointPathFactory() = default; - /// Initializes the counter before the first allocation, without storage I/O. - /// @param last_file_id Largest existing checkpoint id, or -1 if no checkpoint exists. - virtual void InitializeFileId(int64_t last_file_id) = 0; - - /// Allocates the next id and creates its path, without storage I/O. Fails on id overflow. - virtual Result NewPath() const = 0; + /// 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. diff --git a/src/paimon/core/utils/file_store_path_factory.cpp b/src/paimon/core/utils/file_store_path_factory.cpp index b38f530cf..ac15ad638 100644 --- a/src/paimon/core/utils/file_store_path_factory.cpp +++ b/src/paimon/core/utils/file_store_path_factory.cpp @@ -19,7 +19,6 @@ #include "paimon/core/utils/file_store_path_factory.h" #include -#include #include "fmt/format.h" #include "paimon/common/fs/external_path_provider.h" @@ -224,17 +223,10 @@ FileStorePathFactory::CreateGlobalIndexCheckpointPathFactory(const std::string& const std::string& file_name_prefix) : directory_(directory), file_name_prefix_(file_name_prefix) {} - void InitializeFileId(int64_t last_file_id) override { - assert(last_file_id >= -1); - last_file_id_ = last_file_id; - } - - Result NewPath() const override { - if (last_file_id_ == std::numeric_limits::max()) { - return Status::Invalid("checkpoint file id exceeds int64 max"); - } - std::string file_name = fmt::format("{}{}{}", file_name_prefix_, ++last_file_id_, - kIndexCheckpointFileSuffix); + 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); } @@ -253,7 +245,6 @@ FileStorePathFactory::CreateGlobalIndexCheckpointPathFactory(const std::string& private: std::string directory_; std::string file_name_prefix_; - mutable int64_t last_file_id_ = -1; }; PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint index type", index_type)); PAIMON_RETURN_NOT_OK(PathUtil::CheckSinglePathComponent("checkpoint field", field_name)); 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 67bf039c3..bfb2b9621 100644 --- a/src/paimon/core/utils/file_store_path_factory_test.cpp +++ b/src/paimon/core/utils/file_store_path_factory_test.cpp @@ -602,10 +602,9 @@ TEST_F(FileStorePathFactoryTest, TestCreateGlobalIndexCheckpointPathFactory) { "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); - checkpoint_path_factory->InitializeFileId(-1); - ASSERT_OK_AND_ASSIGN(std::string first_path, checkpoint_path_factory->NewPath()); + std::string first_path = checkpoint_path_factory->NewPath(0); ASSERT_EQ(first_path, PathUtil::JoinPath(checkpoint_dir, prefix + "0.index.ckpt")); - ASSERT_OK_AND_ASSIGN(std::string second_path, checkpoint_path_factory->NewPath()); + 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")); @@ -632,26 +631,17 @@ TEST_F(FileStorePathFactoryTest, TestCreateGlobalIndexCheckpointPathFactory) { ASSERT_FALSE(exists); } -TEST_F(FileStorePathFactoryTest, TestCheckpointFileIdInitializationAndOverflow) { +TEST_F(FileStorePathFactoryTest, TestCheckpointFileIdPath) { auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); auto factory = CreateFactory(dir->Str()); - for (int64_t last_id : {int64_t{9}, std::numeric_limits::max() - 1, + for (int64_t file_id : {int64_t{9}, std::numeric_limits::max() - 1, std::numeric_limits::max()}) { ASSERT_OK_AND_ASSIGN(std::unique_ptr checkpoint_factory, factory->CreateGlobalIndexCheckpointPathFactory( "lumina", "vector", Range(10, 20), "task-1")); - checkpoint_factory->InitializeFileId(last_id); - if (last_id != std::numeric_limits::max()) { - ASSERT_OK_AND_ASSIGN(std::string path, checkpoint_factory->NewPath()); - ASSERT_EQ(checkpoint_factory->GetCheckpointId(PathUtil::GetName(path)), last_id + 1); - } - if (last_id >= std::numeric_limits::max() - 1) { - ASSERT_NOK_WITH_MSG(checkpoint_factory->NewPath(), - "checkpoint file id exceeds int64 max"); - ASSERT_NOK_WITH_MSG(checkpoint_factory->NewPath(), - "checkpoint file id exceeds int64 max"); - } + 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); @@ -671,10 +661,9 @@ TEST_F(FileStorePathFactoryTest, TestCheckpointTaskIsolation) { Range(10, 20), "task-2")); ASSERT_EQ(own_factory->GetDirectoryPath(), other_factory->GetDirectoryPath()); ASSERT_OK(fs->Mkdirs(own_factory->GetDirectoryPath())); - ASSERT_OK_AND_ASSIGN(std::string own_path, own_factory->NewPath()); + std::string own_path = own_factory->NewPath(0); ASSERT_OK(fs->WriteFile(own_path, "checkpoint", /*overwrite=*/false)); - other_factory->InitializeFileId(99); - ASSERT_OK_AND_ASSIGN(std::string other_path, other_factory->NewPath()); + 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); 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 1e30b366c..a63e96e66 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(options); auto path_factory = std::make_shared(index_root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr global_writer, global_index->CreateWriter("f0", CreateArrowSchema(data_type).get(), @@ -114,7 +115,8 @@ class LuceneGlobalIndexTest : public ::testing::Test, const std::map& options, const GlobalIndexIOMeta& meta) const { auto global_index = std::make_shared(options); auto path_factory = std::make_shared(index_root); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + 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/lumina_global_index.cpp b/src/paimon/global_index/lumina/lumina_global_index.cpp index 16863f128..06383e9fc 100644 --- a/src/paimon/global_index/lumina/lumina_global_index.cpp +++ b/src/paimon/global_index/lumina/lumina_global_index.cpp @@ -944,18 +944,14 @@ Result> LuminaIndexWriter::Finish() { extension->LoadCkptManager(std::move(checkpoint_manager))); return Status::OK(); }; - Status checkpoint_status = Status::OK(); if (tag_fields_.empty()) { context->checkpoint_extension = std::make_unique< ::lumina::extensions::experimental::BuildWithCheckpointExtension>(); - checkpoint_status = attach_checkpoint(context->checkpoint_extension.get()); + PAIMON_RETURN_NOT_OK(attach_checkpoint(context->checkpoint_extension.get())); } else { context->checkpoint_tag_extension = std::make_unique< ::lumina::extensions::experimental::BuildWithCkptAndTagExtension>(); - checkpoint_status = attach_checkpoint(context->checkpoint_tag_extension.get()); - } - if (!checkpoint_status.ok()) { - return checkpoint_status; + PAIMON_RETURN_NOT_OK(attach_checkpoint(context->checkpoint_tag_extension.get())); } } else if (!tag_fields_.empty()) { context->tag_extension = @@ -1000,8 +996,6 @@ Result> LuminaIndexWriter::Finish() { return build_result.status(); } std::unique_ptr build_context = std::move(build_result).value(); - std::vector>().swap(array_vec_); - std::vector>().swap(tag_data_vec_); // dump index PAIMON_ASSIGN_OR_RAISE(std::string index_file_name, @@ -1020,7 +1014,11 @@ Result> LuminaIndexWriter::Finish() { GlobalIndexIOMeta meta(file_manager_->ToPath(index_file_name), file_size, /*metadata=*/meta_bytes); if (checkpoint_file_manager_) { - PAIMON_RETURN_NOT_OK(checkpoint_file_manager_->DeleteCheckpoint()); + Status status = checkpoint_file_manager_->DeleteCheckpoint(); + if (!status.ok()) { + LOG(WARNING) << "Failed to delete Lumina checkpoints after successful build: " + << status.ToString(); + } } return std::vector({meta}); } 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 e7dc12f26..0ad49a966 100644 --- a/src/paimon/global_index/lumina/lumina_global_index_test.cpp +++ b/src/paimon/global_index/lumina/lumina_global_index_test.cpp @@ -225,7 +225,8 @@ class LuminaGlobalIndexTest : public ::testing::Test { const std::map& options, const GlobalIndexIOMeta& meta) const { auto global_index = std::make_shared(options); auto path_factory = std::make_shared(index_root); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr); return global_index->CreateReader(CreateArrowSchema(data_type).get(), file_reader, {meta}, pool_); } @@ -357,7 +358,8 @@ TEST_F(LuminaGlobalIndexTest, TestCheckpointCapabilityOnlyRequiredWhenEnabled) { ASSERT_OK_AND_ASSIGN(std::shared_ptr path_factory, CreateFileStorePathFactory(dir->Str())); auto plain_manager = - std::make_shared(fs_, path_factory->CreateGlobalIndexFileFactory()); + std::make_shared(fs_, path_factory->CreateGlobalIndexFileFactory(), + /*checkpoint_path_factory=*/nullptr); ASSERT_OK_AND_ASSIGN(std::shared_ptr checkpoint_manager, CreateCountingCheckpointFileManager(path_factory, Range(0, 3))); std::map options = options_; @@ -1219,7 +1221,8 @@ TEST_F(LuminaGlobalIndexTest, TestInvalidInputs) { { auto global_index = std::make_shared(options_); auto path_factory = std::make_shared(index_root); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr); ASSERT_NOK_WITH_MSG(global_index->CreateReader(CreateArrowSchema(data_type_).get(), file_reader, {meta, meta}, pool_), @@ -1437,7 +1440,8 @@ TEST_F(LuminaGlobalIndexTest, TestWriteWithAllNullRows) { auto global_index = std::make_shared(options_); auto path_factory = std::make_shared(test_root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr); ASSERT_OK_AND_ASSIGN( std::shared_ptr global_writer, @@ -1509,7 +1513,8 @@ TEST_F(LuminaGlobalIndexTest, TestWriteWithNullAcrossMultipleBatches) { auto global_index = std::make_shared(options_); auto path_factory = std::make_shared(test_root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr); ASSERT_OK_AND_ASSIGN( std::shared_ptr 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 44f780ac3..07f06200c 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(root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + 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(root); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + 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 5de18683b..19df5ac64 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(root); - auto fm = std::make_shared(fs_, path_factory); + auto fm = std::make_shared(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 199e347b5..d901d77c6 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(options); auto path_factory = std::make_shared(root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr w, global_index->CreateWriter("f0", CreateArrowSchema(data_type).get(), file_writer, pool_)); @@ -111,7 +112,8 @@ class TantivyGlobalIndexIntegrationTest : public ::testing::Test { const std::map& options, const GlobalIndexIOMeta& meta) const { auto global_index = std::make_shared(options); auto path_factory = std::make_shared(root); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + 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 b998402ce..c3358482e 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 options; auto global_index = std::make_shared(options); auto path_factory = std::make_shared(fixture_dir); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + 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(std::map{}); auto reader_path_factory = std::make_shared(out_dir); - auto reader_file_mgr = std::make_shared(fs_, reader_path_factory); + auto reader_file_mgr = std::make_shared( + 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 2b6fdeb5f..e837a0de3 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(root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr 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 indexer, GlobalIndexerFactory::Get(impl.factory_id, options)); auto path_factory = std::make_shared(root); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + 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 36ca7a3a2..d5de32654 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(options); auto path_factory = std::make_shared(root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + 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(options); auto path_factory = std::make_shared(root); - auto file_reader = std::make_shared(fs_, path_factory); + auto file_reader = std::make_shared( + 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 0d623b889..886f7f5c7 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& options, const std::shared_ptr& array) { auto path_factory = std::make_shared(root); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + 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(root_dir->Str()); - auto file_writer = std::make_shared(fs_, path_factory); + auto file_writer = std::make_shared( + 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). From 306b02da6f77862657b0951b94450a3106de1cad Mon Sep 17 00:00:00 2001 From: "lisizhuo.lsz" Date: Wed, 23 Sep 2026 21:52:50 +0800 Subject: [PATCH 4/4] fix rebase --- src/paimon/core/index/pksorted/pk_sorted_index_builder.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 765c1ea3c..4d36aa1f3 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> PkSortedIndexBuilder::Build( auto sorted_reader = std::make_unique( std::move(readers), comparator, sequence_comparator, /*merge_function_wrapper=*/nullptr); - auto file_manager = std::make_shared(fs_, index_path_factory_); + auto file_manager = std::make_shared( + fs_, index_path_factory_, /*checkpoint_path_factory=*/nullptr); auto tracking_writer = std::make_shared(file_manager); Result> result = PkSortedIndexFile::BuildFromSortedReader( field_, definition_.IndexType(), definition_.Options(), data_level, source_metas,