test

Author: sollhui

be/src/common/config.cpp

@@ -726,6 +726,8 @@ DEFINE_mInt32(memory_gc_sleep_time_ms, "500");
 // max write buffer size before flush, default 200MB
 DEFINE_mInt64(write_buffer_size, "209715200");
 DEFINE_mBool(enable_adaptive_write_buffer_size, "true");
+// Whether random bucket load rotates to the next local bucket when memtable flushes.
+DEFINE_mBool(enable_adaptive_random_bucket_load_bucket_rotation, "true");
 // max buffer size used in memtable for the aggregated table, default 400MB
 DEFINE_mInt64(write_buffer_size_for_agg, "104857600");
 DEFINE_mInt64(min_write_buffer_size_for_partial_update, "1048576");

be/src/common/config.h

@@ -785,6 +785,8 @@ DECLARE_mInt32(memory_gc_sleep_time_ms);
 // max write buffer size before flush, default 200MB
 DECLARE_mInt64(write_buffer_size);
 DECLARE_mBool(enable_adaptive_write_buffer_size);
+// Whether random bucket load rotates to the next local bucket when memtable flushes.
+DECLARE_mBool(enable_adaptive_random_bucket_load_bucket_rotation);
 // max buffer size used in memtable for the aggregated table, default 400MB
 DECLARE_mInt64(write_buffer_size_for_agg);
 

be/src/exec/sink/vtablet_finder.cpp

@@ -27,6 +27,7 @@
 #include <utility>
 
 #include "common/compiler_util.h" // IWYU pragma: keep
+#include "common/config.h"
 #include "common/status.h"
 #include "core/block/block.h"
 #include "runtime/runtime_state.h"
@@ -57,7 +58,7 @@ std::shared_ptr<AdaptiveRandomBucketState> AdaptiveRandomBucketStatePool::get_or
     if (auto state = _pool[load_id].lock()) {
         return state;
     }
-    auto state = std::make_shared<AdaptiveRandomBucketState>();
+    auto state = std::make_shared<AdaptiveRandomBucketState>(load_id);
     _pool[load_id] = state;
     return state;
 }
@@ -97,7 +98,8 @@ void AdaptiveRandomBucketState::init_partition(VOlapTablePartition* part) {
         }
     }
     _partition_states.emplace(part->id, std::move(state));
-    LOG(INFO) << "FIND_TABLET_RANDOM_BUCKET: partition=" << part->id
+    LOG(INFO) << "FIND_TABLET_RANDOM_BUCKET: load_id=" << _load_id
+              << ", partition=" << part->id
               << " shared start bucket=" << current_bucket
               << ", local_bucket_seqs=" << format_bucket_seqs(part->local_bucket_seqs);
 }
@@ -112,6 +114,9 @@ int64_t AdaptiveRandomBucketState::current_bucket(int64_t partition_id) {
 }
 
 void AdaptiveRandomBucketState::rotate_by_tablet(int64_t tablet_id) {
+    if (!config::enable_adaptive_random_bucket_load_bucket_rotation) {
+        return;
+    }
     std::lock_guard<std::mutex> lock(_mutex);
     auto partition_it = _tablet_to_partition.find(tablet_id);
     if (partition_it == _tablet_to_partition.end()) {
@@ -132,7 +137,8 @@ void AdaptiveRandomBucketState::rotate_by_tablet(int64_t tablet_id) {
     }
     int32_t next_pos = (state.seq_pos + 1) % static_cast<int32_t>(state.local_bucket_seqs.size());
     int32_t next_bucket = state.local_bucket_seqs[next_pos];
-    LOG(INFO) << "FIND_TABLET_RANDOM_BUCKET: partition=" << state.partition_id
+    LOG(INFO) << "FIND_TABLET_RANDOM_BUCKET: load_id=" << _load_id
+              << ", partition=" << state.partition_id
               << " shared rotate bucket " << current_bucket << " -> " << next_bucket
               << " after tablet=" << tablet_id << " memtable flushed";
     state.seq_pos = next_pos;

be/src/exec/sink/vtablet_finder.h

@@ -43,6 +43,8 @@ void compute_local_bucket_seqs(VOlapTablePartition* part, int64_t be_id,
 
 class AdaptiveRandomBucketState {
 public:
+    explicit AdaptiveRandomBucketState(UniqueId load_id) : _load_id(load_id) {}
+
     void init_partition(VOlapTablePartition* part);
     int64_t current_bucket(int64_t partition_id);
     void rotate_by_tablet(int64_t tablet_id);
@@ -57,6 +59,7 @@ class AdaptiveRandomBucketState {
     };
 
     std::mutex _mutex;
+    UniqueId _load_id;
     std::unordered_map<int64_t, PartitionState> _partition_states;
     std::unordered_map<int64_t, int64_t> _tablet_to_partition;
     std::unordered_map<int64_t, int32_t> _tablet_to_bucket;