adaptive random bucket load: per-BE bucket assignment with threshold-based rotation

Author: sollhui

be/src/exec/sink/vtablet_finder.cpp

@@ -120,9 +120,22 @@ Status OlapTabletFinder::find_tablets(RuntimeState* state, Block* block, int row
                 }
             }
 
-            int64_t cur_bucket = seqs.empty()
-                                         ? partition->load_tablet_idx % partition->num_buckets
-                                         : seqs[seq_pos];
+            // Determine the current bucket:
+            //   - adaptive mode: use the cached position in local_bucket_seqs (O(1))
+            //   - fallback (no seqs, or dynamically added partition without seqs): use
+            //     load_tablet_idx if valid, otherwise fall back to fast_rand() to match
+            //     the FIND_TABLET_EVERY_BATCH behaviour for partitions that were added
+            //     at runtime (e.g. auto-partition, overwrite replace) and did not go
+            //     through assignRandomBucketPerBe.
+            int64_t cur_bucket;
+            if (!seqs.empty()) {
+                cur_bucket = seqs[seq_pos];
+            } else if (partition->load_tablet_idx >= 0) {
+                cur_bucket = partition->load_tablet_idx % partition->num_buckets;
+            } else {
+                cur_bucket =
+                        static_cast<int64_t>(butil::fast_rand()) % partition->num_buckets;
+            }
             int64_t cur_tablet_id = partition->indexes[0].tablets[cur_bucket];
 
             // Credit only the rows that actually went to this partition.