fix(lucene): isolate lock factories for concurrent directory creation

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::NoLockFactory>());
 }
 
 Lucene::HashSet<Lucene::String> LuceneDirectory::listAll() {

src/paimon/global_index/lucene/lucene_directory_test.cpp

@@ -17,13 +17,19 @@
  */
 #include "paimon/global_index/lucene/lucene_directory.h"
 
+#include <future>
+#include <set>
+#include <thread>
+
+#include "fmt/format.h"
 #include "gtest/gtest.h"
 #include "lucene++/FileUtils.h"
 #include "lucene++/LuceneHeaders.h"
 #include "lucene++/MiscUtils.h"
 #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<int32_t> {
@@ -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<std::unique_ptr<LuceneDirectory>> directories(kThreadCount);
+    std::vector<std::promise<void>> ready_signals(kThreadCount);
+    std::promise<void> start_signal;
+    std::shared_future<void> start_future = start_signal.get_future().share();
+    std::vector<std::thread> threads;
+    threads.reserve(kThreadCount);
+    for (int32_t i = 0; i < kThreadCount; ++i) {
+        threads.emplace_back([&, i]() {
+            auto input = std::make_shared<paimon::test::MockInputStream>();
+            std::map<std::string, std::pair<int64_t, int64_t>> files;
+            ready_signals[i].set_value();
+            start_future.wait();
+            directories[i] = std::make_unique<LuceneDirectory>(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<Lucene::String> 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

src/paimon/global_index/lucene/lucene_global_index_test.cpp

@@ -18,10 +18,15 @@
 #include "paimon/global_index/lucene/lucene_global_index.h"
 
 #include <cmath>
+#include <future>
+#include <set>
+#include <thread>
 
 #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<std::string, std::string> options = {{kLuceneWriteTmpDir, tmp_dir->Str()}};
+    std::vector<std::shared_ptr<GlobalIndexFileWriter>> 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<FakeIndexPathFactory>(index_root);
+        file_writers.push_back(std::make_shared<GlobalIndexFileManager>(
+            fs_, path_factory, /*checkpoint_path_factory=*/nullptr));
+    }
+
+    std::vector<std::shared_ptr<LuceneGlobalIndexWriter>> writers(kThreadCount);
+    std::vector<std::promise<void>> ready_signals(kThreadCount);
+    std::promise<void> start_signal;
+    std::shared_future<void> start_future = start_signal.get_future().share();
+    std::vector<std::thread> 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<Lucene::String> 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();

src/paimon/global_index/lucene/lucene_global_index_writer.cpp

@@ -65,8 +65,10 @@ Result<std::shared_ptr<LuceneGlobalIndexWriter>> 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<Lucene::NoLockFactory>());
         // TODO(xinyu.lxy): support other tokenizer
         // open lucene index writer
         PAIMON_ASSIGN_OR_RAISE(