fix: fix IsThreadSafe() in UnionGlobalIndexReader (alibaba#395)

Author: lszskye

src/paimon/common/global_index/union_global_index_reader.cpp

@@ -182,6 +182,15 @@ Result<std::shared_ptr<GlobalIndexResult>> UnionGlobalIndexReader::Union(ReaderA
     return merged_result;
 }
 
+bool UnionGlobalIndexReader::IsThreadSafe() const {
+    for (const auto& reader : readers_) {
+        if (!reader->IsThreadSafe()) {
+            return false;
+        }
+    }
+    return true;
+}
+
 template <typename R>
 std::vector<R> UnionGlobalIndexReader::ExecuteAllReaders(
     const std::function<R(const std::shared_ptr<GlobalIndexReader>&)>& action) {
@@ -199,7 +208,7 @@ std::vector<R> UnionGlobalIndexReader::ExecuteAllReaders(
     futures.reserve(readers_.size());
     for (const auto& reader : readers_) {
         futures.push_back(
-            Via(executor_.get(), [&action, &reader]() -> R { return action(reader); }));
+            Via(executor_.get(), [&action, reader]() -> R { return action(reader); }));
     }
     return CollectAll(futures);
 }

src/paimon/common/global_index/union_global_index_reader.h

@@ -58,9 +58,7 @@ class UnionGlobalIndexReader : public GlobalIndexReader {
     Result<std::shared_ptr<GlobalIndexResult>> VisitFullTextSearch(
         const std::shared_ptr<FullTextSearch>& full_text_search) override;
 
-    bool IsThreadSafe() const override {
-        return false;
-    }
+    bool IsThreadSafe() const override;
 
     std::string GetIndexType() const override {
         return "union";

src/paimon/common/global_index/union_global_index_reader_test.cpp

@@ -61,6 +61,10 @@ class FakeReader : public GlobalIndexReader {
         has_scored_result_ = true;
     }
 
+    void SetThreadSafe(bool thread_safe) {
+        thread_safe_ = thread_safe;
+    }
+
     /// Counts how many times any Visit* method was invoked. Useful to assert all readers
     /// are exercised by UnionGlobalIndexReader.
     int InvocationCount() const {
@@ -147,7 +151,7 @@ class FakeReader : public GlobalIndexReader {
     }
 
     bool IsThreadSafe() const override {
-        return true;
+        return thread_safe_;
     }
 
     std::string GetIndexType() const override {
@@ -176,6 +180,7 @@ class FakeReader : public GlobalIndexReader {
     std::vector<int64_t> scored_row_ids_;
     std::vector<float> scored_scores_;
     bool has_scored_result_ = false;
+    bool thread_safe_ = true;
     std::atomic<int32_t> invocation_count_{0};
 };
 
@@ -515,12 +520,24 @@ TEST_F(UnionGlobalIndexReaderTest, TestVisitVectorSearchErrorPropagation) {
     ASSERT_NOK_WITH_MSG(union_reader.VisitVectorSearch(nullptr), "vector search failure");
 }
 
-TEST_F(UnionGlobalIndexReaderTest, TestIsThreadSafeAlwaysFalse) {
-    auto reader = std::make_shared<FakeReader>();
-    std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader};
+TEST_F(UnionGlobalIndexReaderTest, TestIsThreadSafeReturnsTrueWhenAllReadersAreSafe) {
+    auto reader1 = std::make_shared<FakeReader>();
+    auto reader2 = std::make_shared<FakeReader>();
+
+    std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1, reader2};
+    UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
+
+    ASSERT_TRUE(union_reader.IsThreadSafe());
+}
+
+TEST_F(UnionGlobalIndexReaderTest, TestIsThreadSafeReturnsFalseWhenAnyReaderIsNotSafe) {
+    auto reader1 = std::make_shared<FakeReader>();
+    auto reader2 = std::make_shared<FakeReader>();
+    reader2->SetThreadSafe(false);
+
+    std::vector<std::shared_ptr<GlobalIndexReader>> readers = {reader1, reader2};
     UnionGlobalIndexReader union_reader(std::move(readers), nullptr);
 
-    // UnionGlobalIndexReader is not thread-safe regardless of inner readers
     ASSERT_FALSE(union_reader.IsThreadSafe());
 }