[fix](scan) Fix missing sync rowsets in cloud mode (#31717)

apache · Mar 4, 2024 · fb2c401 · fb2c401
1 parent c90728c
commit fb2c401
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 11 deletions.
diff --git a/be/src/cloud/cloud_meta_mgr.cpp b/be/src/cloud/cloud_meta_mgr.cpp
@@ -74,10 +74,7 @@ Status bthread_fork_join(const std::vector<std::function<Status()>>& tasks, int
         return nullptr;
     };
 
-    std::vector<bthread_t> bthread_ids;
-    bthread_ids.resize(tasks.size());
-    for (int task_idx = 0; task_idx < tasks.size(); ++task_idx) {
-        auto* task = &(tasks[task_idx]);
+    for (const auto& task : tasks) {
         {
             std::unique_lock lk(lock);
             // Wait until there are available slots
@@ -93,8 +90,8 @@ Status bthread_fork_join(const std::vector<std::function<Status()>>& tasks, int
         }
 
         // dispatch task into bthreads
-        auto* fn = new std::function<void()>([&, task] {
-            auto st = (*task)();
+        auto* fn = new std::function<void()>([&, &task = task] {
+            auto st = task();
             {
                 std::lock_guard lk(lock);
                 --count;
@@ -104,7 +101,9 @@ Status bthread_fork_join(const std::vector<std::function<Status()>>& tasks, int
                 cond.notify_one();
             }
         });
-        if (bthread_start_background(&bthread_ids[task_idx], nullptr, run_bthread_work, fn) != 0) {
+
+        bthread_t bthread_id;
+        if (bthread_start_background(&bthread_id, nullptr, run_bthread_work, fn) != 0) {
             run_bthread_work(fn);
         }
     }

diff --git a/be/src/common/sync_point.h b/be/src/common/sync_point.h
@@ -155,15 +155,15 @@ auto try_any_cast_ret(std::vector<std::any>& any) {
 { \
   std::pair ret {default_ret_val, false}; \
   std::vector<std::any> args {__VA_ARGS__}; \
-  args.push_back(&ret); \
+  args.emplace_back(&ret); \
   doris::SyncPoint::get_instance()->process(x, std::move(args)); \
   if (ret.second) return std::move(ret.first); \
 }
 #define SYNC_POINT_RETURN_WITH_VOID(x, ...) \
 { \
   bool pred = false; \
   std::vector<std::any> args {__VA_ARGS__}; \
-  args.push_back(&pred); \
+  args.emplace_back(&pred); \
   doris::SyncPoint::get_instance()->process(x, std::move(args)); \
   if (pred) return; \
 }

diff --git a/be/src/olap/parallel_scanner_builder.cpp b/be/src/olap/parallel_scanner_builder.cpp
@@ -17,6 +17,9 @@
 
 #include "parallel_scanner_builder.h"
 
+#include "cloud/cloud_meta_mgr.h"
+#include "cloud/cloud_tablet.h"
+#include "cloud/config.h"
 #include "olap/rowset/beta_rowset.h"
 #include "pipeline/exec/olap_scan_operator.h"
 #include "vec/exec/scan/new_olap_scanner.h"
@@ -40,6 +43,18 @@ template <typename ParentType>
 Status ParallelScannerBuilder<ParentType>::_build_scanners_by_rowid(
         std::list<VScannerSPtr>& scanners) {
     DCHECK_GE(_rows_per_scanner, _min_rows_per_scanner);
+
+    if (config::is_cloud_mode()) {
+        std::vector<std::function<Status()>> tasks;
+        tasks.reserve(_tablets.size());
+        for (auto&& [tablet, version] : _tablets) {
+            tasks.emplace_back([tablet, version]() {
+                return std::dynamic_pointer_cast<CloudTablet>(tablet)->sync_rowsets(version);
+            });
+        }
+        RETURN_IF_ERROR(cloud::bthread_fork_join(tasks, 10));
+    }
+
     for (auto&& [tablet, version] : _tablets) {
         DCHECK(_all_rowsets.contains(tablet->tablet_id()));
         auto& rowsets = _all_rowsets[tablet->tablet_id()];

diff --git a/be/src/pipeline/exec/olap_scan_operator.cpp b/be/src/pipeline/exec/olap_scan_operator.cpp
@@ -21,6 +21,9 @@
 
 #include <memory>
 
+#include "cloud/cloud_meta_mgr.h"
+#include "cloud/cloud_tablet.h"
+#include "cloud/config.h"
 #include "olap/parallel_scanner_builder.h"
 #include "olap/storage_engine.h"
 #include "olap/tablet_manager.h"
@@ -277,8 +280,18 @@ Status OlapScanLocalState::_init_scanners(std::list<vectorized::VScannerSPtr>* s
             int64_t version = 0;
             std::from_chars(scan_range->version.data(),
                             scan_range->version.data() + scan_range->version.size(), version);
-            tablets.emplace_back(
-                    TabletWithVersion {std::dynamic_pointer_cast<Tablet>(tablet), version});
+            tablets.emplace_back(TabletWithVersion {std::move(tablet), version});
+        }
+
+        if (config::is_cloud_mode()) {
+            std::vector<std::function<Status()>> tasks;
+            tasks.reserve(tablets.size());
+            for (auto&& [tablet, version] : tablets) {
+                tasks.emplace_back([tablet, version]() {
+                    return std::dynamic_pointer_cast<CloudTablet>(tablet)->sync_rowsets(version);
+                });
+            }
+            RETURN_IF_ERROR(cloud::bthread_fork_join(tasks, 10));
         }
 
         if (is_dup_mow_key) {