diff --git a/src/paimon/global_index/lucene/lucene_directory.cpp b/src/paimon/global_index/lucene/lucene_directory.cpp index 5ea7d91a4..b8cccb794 100644 --- a/src/paimon/global_index/lucene/lucene_directory.cpp +++ b/src/paimon/global_index/lucene/lucene_directory.cpp @@ -40,7 +40,9 @@ LuceneDirectory::LuceneDirectory( path_(path), file_name_to_offset_and_length_(file_name_to_offset_and_length), paimon_input_(paimon_input) { - Lucene::Directory::setLockFactory(Lucene::NoLockFactory::getNoLockFactory()); + // Directory::setLockFactory() mutates lockPrefix. Do not share the NoLockFactory singleton + // across concurrently created directories, even though the locks themselves are no-ops. + Lucene::Directory::setLockFactory(Lucene::newLucene()); } Lucene::HashSet LuceneDirectory::listAll() { diff --git a/src/paimon/global_index/lucene/lucene_directory_test.cpp b/src/paimon/global_index/lucene/lucene_directory_test.cpp index ab15ed43a..125bf656c 100644 --- a/src/paimon/global_index/lucene/lucene_directory_test.cpp +++ b/src/paimon/global_index/lucene/lucene_directory_test.cpp @@ -17,6 +17,11 @@ */ #include "paimon/global_index/lucene/lucene_directory.h" +#include +#include +#include + +#include "fmt/format.h" #include "gtest/gtest.h" #include "lucene++/FileUtils.h" #include "lucene++/LuceneHeaders.h" @@ -24,6 +29,7 @@ #include "paimon/common/utils/path_util.h" #include "paimon/fs/local/local_file_system.h" #include "paimon/global_index/lucene/lucene_utils.h" +#include "paimon/testing/mock/mock_file_system.h" #include "paimon/testing/utils/testharness.h" namespace paimon::lucene::test { class LuceneDirectoryTest : public ::testing::Test, public ::testing::WithParamInterface { @@ -32,6 +38,42 @@ class LuceneDirectoryTest : public ::testing::Test, public ::testing::WithParamI void TearDown() override {} }; +TEST_P(LuceneDirectoryTest, TestLockFactoryIsolation) { + constexpr int32_t kThreadCount = 8; + int32_t read_buffer_size = GetParam(); + std::vector> directories(kThreadCount); + std::vector> ready_signals(kThreadCount); + std::promise start_signal; + std::shared_future start_future = start_signal.get_future().share(); + std::vector threads; + threads.reserve(kThreadCount); + for (int32_t i = 0; i < kThreadCount; ++i) { + threads.emplace_back([&, i]() { + auto input = std::make_shared(); + std::map> files; + ready_signals[i].set_value(); + start_future.wait(); + directories[i] = std::make_unique(fmt::format("directory-{}", i), + files, input, read_buffer_size); + }); + } + for (auto& ready_signal : ready_signals) { + ready_signal.get_future().wait(); + } + start_signal.set_value(); + for (auto& thread : threads) { + thread.join(); + } + + std::set lock_prefixes; + for (const auto& directory : directories) { + ASSERT_TRUE(directory); + auto prefix = directory->getLockFactory()->getLockPrefix(); + ASSERT_EQ(prefix, directory->getLockID()); + ASSERT_TRUE(lock_prefixes.insert(prefix).second); + } +} + TEST_P(LuceneDirectoryTest, TestSimple) { int32_t read_buffer_size = GetParam(); // write 3 files in a single concat file 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 905c345c7..273bd19e6 100644 --- a/src/paimon/global_index/lucene/lucene_global_index_test.cpp +++ b/src/paimon/global_index/lucene/lucene_global_index_test.cpp @@ -18,10 +18,15 @@ #include "paimon/global_index/lucene/lucene_global_index.h" #include +#include +#include +#include #include "arrow/c/bridge.h" #include "arrow/ipc/api.h" +#include "fmt/format.h" #include "gtest/gtest.h" +#include "lucene++/LockFactory.h" #include "paimon/common/utils/arrow/status_utils.h" #include "paimon/common/utils/path_util.h" #include "paimon/common/utils/string_utils.h" @@ -152,6 +157,57 @@ class LuceneGlobalIndexTest : public ::testing::Test, arrow::struct_({arrow::field("f0", arrow::utf8())}); }; +TEST_F(LuceneGlobalIndexTest, TestConcurrentWriterCreation) { + constexpr int32_t kThreadCount = 8; + auto test_root_dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(test_root_dir); + auto tmp_dir = paimon::test::UniqueTestDirectory::Create(); + ASSERT_TRUE(tmp_dir); + std::map options = {{kLuceneWriteTmpDir, tmp_dir->Str()}}; + std::vector> file_writers; + file_writers.reserve(kThreadCount); + for (int32_t i = 0; i < kThreadCount; ++i) { + auto index_root = PathUtil::JoinPath(test_root_dir->Str(), fmt::format("writer-{}", i)); + ASSERT_OK(fs_->Mkdirs(index_root)); + auto path_factory = std::make_shared(index_root); + file_writers.push_back(std::make_shared( + fs_, path_factory, /*checkpoint_path_factory=*/nullptr)); + } + + std::vector> writers(kThreadCount); + std::vector> ready_signals(kThreadCount); + std::promise start_signal; + std::shared_future start_future = start_signal.get_future().share(); + std::vector threads; + threads.reserve(kThreadCount); + for (int32_t i = 0; i < kThreadCount; ++i) { + threads.emplace_back([&, i]() { + ready_signals[i].set_value(); + start_future.wait(); + ASSERT_OK_AND_ASSIGN( + writers[i], + LuceneGlobalIndexWriter::Create("f0", data_type_, file_writers[i], options, pool_)); + }); + } + for (auto& ready_signal : ready_signals) { + ready_signal.get_future().wait(); + } + start_signal.set_value(); + for (auto& thread : threads) { + thread.join(); + } + + // Keep all writers alive so sharing a factory fails even if the race did not crash. + std::set lock_prefixes; + for (const auto& writer : writers) { + ASSERT_TRUE(writer); + const auto& directory = writer->write_context_.lucene_dir; + auto prefix = directory->getLockFactory()->getLockPrefix(); + ASSERT_EQ(prefix, directory->getLockID()); + ASSERT_TRUE(lock_prefixes.insert(prefix).second); + } +} + TEST_P(LuceneGlobalIndexTest, TestSimple) { int32_t read_buffer_size = GetParam(); auto test_root_dir = paimon::test::UniqueTestDirectory::Create(); diff --git a/src/paimon/global_index/lucene/lucene_global_index_writer.cpp b/src/paimon/global_index/lucene/lucene_global_index_writer.cpp index 7477302c1..8560746a0 100644 --- a/src/paimon/global_index/lucene/lucene_global_index_writer.cpp +++ b/src/paimon/global_index/lucene/lucene_global_index_writer.cpp @@ -65,8 +65,10 @@ Result> LuceneGlobalIndexWriter::Create options, std::string(kLuceneWriteTmpDir))); std::string tmp_path = PathUtil::JoinPath(tmp_dir, "paimon-lucene-" + uuid); + // FSDirectory calls Directory::setLockFactory(), which mutates lockPrefix. Use a separate + // factory per directory to avoid racing on the NoLockFactory singleton across writers. auto lucene_dir = Lucene::FSDirectory::open(LuceneUtils::StringToWstring(tmp_path), - Lucene::NoLockFactory::getNoLockFactory()); + Lucene::newLucene()); // TODO(xinyu.lxy): support other tokenizer // open lucene index writer PAIMON_ASSIGN_OR_RAISE(