[Enhancement] [Storage Vectorize] optimize BitmapRangeIterator.next_r…

…ange() (apache#9013)

be/src/olap/rowset/segment_v2/segment_iterator.cpp

@@ -57,23 +57,38 @@ class SegmentIterator::BitmapRangeIterator {
 
     // read next range into [*from, *to) whose size <= max_range_size.
     // return false when there is no more range.
-    bool next_range(uint32_t max_range_size, uint32_t* from, uint32_t* to) {
+    bool next_range(const uint32_t max_range_size, uint32_t* from, uint32_t* to) {
         if (_eof) {
             return false;
         }
+
         *from = _buf[_buf_pos];
-        uint32_t range_size = 0, last_val;
-        do {
-            last_val = _buf[_buf_pos];
-            _buf_pos++;
-            range_size++;
-            if (UNLIKELY(_buf_pos == _buf_size)) { // read next batch
-                _read_next_batch();
-                if (_eof) {
-                    break;
-                }
-            }
-        } while (range_size < max_range_size && _buf[_buf_pos] == last_val + 1);
+        uint32_t range_size = 0;
+        uint32_t expect_val = _buf[_buf_pos]; // this initial value just make first batch valid
+
+        // if array is contiguous sequence then the following conditions need to be met :
+        // a_0: x
+        // a_1: x+1
+        // a_2: x+2
+        // ...
+        // a_p: x+p
+        // so we can just use (a_p-a_0)-p to check conditions
+        // and should notice the previous batch needs to be continuous with the current batch
+        while (!_eof && range_size + _buf_size - _buf_pos <= max_range_size &&
+               expect_val == _buf[_buf_pos] &&
+               _buf[_buf_size - 1] - _buf[_buf_pos] == _buf_size - 1 - _buf_pos) {
+            range_size += _buf_size - _buf_pos;
+            expect_val = _buf[_buf_size - 1] + 1;
+            _read_next_batch();
+        }
+
+        // promise remain range not will reach next batch
+        if (!_eof && range_size < max_range_size && expect_val == _buf[_buf_pos]) {
+            do {
+                _buf_pos++;
+                range_size++;
+            } while (range_size < max_range_size && _buf[_buf_pos] == _buf[_buf_pos - 1] + 1);
+        }
         *to = *from + range_size;
         return true;
     }