remove Block::row_same_bit

Author: BiteTheDDDDt

be/src/olap/iterators.h

@@ -159,6 +159,13 @@ struct CompactionSampleInfo {
     int64_t group_data_size;
 };
 
+struct BlockWithSameBit {
+    vectorized::Block* block;
+    std::vector<bool>& same_bit;
+
+    bool empty() const { return block->rows() == 0; }
+};
+
 class RowwiseIterator;
 using RowwiseIteratorUPtr = std::unique_ptr<RowwiseIterator>;
 class RowwiseIterator {
@@ -187,7 +194,11 @@ class RowwiseIterator {
         return Status::NotSupported("to be implemented");
     }
 
-    virtual Status next_block_view(vectorized::BlockView* block_view) {
+    virtual Status next_batch(BlockWithSameBit* block_with_same_bit) {
+        return Status::NotSupported("to be implemented");
+    }
+
+    virtual Status next_batch(vectorized::BlockView* block_view) {
         return Status::NotSupported("to be implemented");
     }
 

be/src/olap/rowset/beta_rowset_reader.cpp

@@ -348,60 +348,6 @@ Status BetaRowsetReader::_init_iterator() {
     return Status::OK();
 }
 
-Status BetaRowsetReader::next_block(vectorized::Block* block) {
-    RETURN_IF_ERROR(_init_iterator_once());
-    SCOPED_RAW_TIMER(&_stats->block_fetch_ns);
-    if (_empty) {
-        return Status::Error<END_OF_FILE>("BetaRowsetReader is empty");
-    }
-
-    RuntimeState* runtime_state = nullptr;
-    if (_read_context != nullptr) {
-        runtime_state = _read_context->runtime_state;
-    }
-
-    do {
-        auto s = _iterator->next_batch(block);
-        if (!s.ok()) {
-            if (!s.is<END_OF_FILE>()) {
-                LOG(WARNING) << "failed to read next block: " << s.to_string();
-            }
-            return s;
-        }
-
-        if (runtime_state != nullptr && runtime_state->is_cancelled()) [[unlikely]] {
-            return runtime_state->cancel_reason();
-        }
-    } while (block->empty());
-
-    return Status::OK();
-}
-
-Status BetaRowsetReader::next_block_view(vectorized::BlockView* block_view) {
-    RETURN_IF_ERROR(_init_iterator_once());
-    SCOPED_RAW_TIMER(&_stats->block_fetch_ns);
-    RuntimeState* runtime_state = nullptr;
-    if (_read_context != nullptr) {
-        runtime_state = _read_context->runtime_state;
-    }
-
-    do {
-        auto s = _iterator->next_block_view(block_view);
-        if (!s.ok()) {
-            if (!s.is<END_OF_FILE>()) {
-                LOG(WARNING) << "failed to read next block view: " << s.to_string();
-            }
-            return s;
-        }
-
-        if (runtime_state != nullptr && runtime_state->is_cancelled()) [[unlikely]] {
-            return runtime_state->cancel_reason();
-        }
-    } while (block_view->empty());
-
-    return Status::OK();
-}
-
 bool BetaRowsetReader::_should_push_down_value_predicates() const {
     // if unique table with rowset [0-x] or [0-1] [2-y] [...],
     // value column predicates can be pushdown on rowset [0-x] or [2-y], [2-y]

be/src/olap/rowset/beta_rowset_reader.h

@@ -53,8 +53,13 @@ class BetaRowsetReader : public RowsetReader {
                                  std::vector<RowwiseIteratorUPtr>* out_iters,
                                  bool use_cache = false) override;
     void reset_read_options() override;
-    Status next_block(vectorized::Block* block) override;
-    Status next_block_view(vectorized::BlockView* block_view) override;
+    Status next_batch(vectorized::Block* block) override { return _next_batch(block); }
+    Status next_batch(vectorized::BlockView* block_view) override {
+        return _next_batch(block_view);
+    }
+    Status next_batch(BlockWithSameBit* block_with_same_bit) override {
+        return _next_batch(block_with_same_bit);
+    }
     bool support_return_data_by_ref() override { return _is_merge_iterator(); }
 
     bool delete_flag() override { return _rowset->delete_flag(); }
@@ -89,6 +94,36 @@ class BetaRowsetReader : public RowsetReader {
     OlapReaderStatistics* get_stats() { return _stats; }
 
 private:
+    template <typename T>
+    Status _next_batch(T* block) {
+        RETURN_IF_ERROR(_init_iterator_once());
+        SCOPED_RAW_TIMER(&_stats->block_fetch_ns);
+        if (_empty) {
+            return Status::Error<ErrorCode::END_OF_FILE>("BetaRowsetReader is empty");
+        }
+
+        RuntimeState* runtime_state = nullptr;
+        if (_read_context != nullptr) {
+            runtime_state = _read_context->runtime_state;
+        }
+
+        do {
+            Status s = _iterator->next_batch(block);
+            if (!s.ok()) {
+                if (!s.is<ErrorCode::END_OF_FILE>()) {
+                    LOG(WARNING) << "failed to read next block: " << s.to_string();
+                }
+                return s;
+            }
+
+            if (runtime_state != nullptr && runtime_state->is_cancelled()) [[unlikely]] {
+                return runtime_state->cancel_reason();
+            }
+        } while (block->empty());
+
+        return Status::OK();
+    }
+
     [[nodiscard]] Status _init_iterator_once();
     [[nodiscard]] Status _init_iterator();
     bool _should_push_down_value_predicates() const;

be/src/olap/rowset/rowset_reader.h

@@ -61,9 +61,9 @@ class RowsetReader {
                                          bool use_cache = false) = 0;
     virtual void reset_read_options() = 0;
 
-    virtual Status next_block(vectorized::Block* block) = 0;
-
-    virtual Status next_block_view(vectorized::BlockView* block_view) = 0;
+    virtual Status next_batch(vectorized::Block* block) = 0;
+    virtual Status next_batch(vectorized::BlockView* block_view) = 0;
+    virtual Status next_batch(BlockWithSameBit* block_view) = 0;
     virtual bool support_return_data_by_ref() { return false; }
 
     virtual bool delete_flag() = 0;

be/src/olap/schema_change.cpp

@@ -554,7 +554,7 @@ Status VSchemaChangeDirectly::_inner_process(RowsetReaderSharedPtr rowset_reader
         auto new_block = vectorized::Block::create_unique(new_tablet_schema->create_block());
         auto ref_block = vectorized::Block::create_unique(base_tablet_schema->create_block());
 
-        auto st = rowset_reader->next_block(ref_block.get());
+        auto st = rowset_reader->next_batch(ref_block.get());
         if (!st) {
             if (st.is<ErrorCode::END_OF_FILE>()) {
                 if (ref_block->rows() == 0) {
@@ -622,7 +622,7 @@ Status VBaseSchemaChangeWithSorting::_inner_process(RowsetReaderSharedPtr rowset
     bool eof = false;
     do {
         auto ref_block = vectorized::Block::create_unique(base_tablet_schema->create_block());
-        auto st = rowset_reader->next_block(ref_block.get());
+        auto st = rowset_reader->next_batch(ref_block.get());
         if (!st) {
             if (st.is<ErrorCode::END_OF_FILE>()) {
                 if (ref_block->rows() == 0) {

be/src/vec/core/block.cpp

@@ -244,9 +244,6 @@ void Block::erase_tail(size_t start) {
             ++it;
         }
     }
-    if (start < row_same_bit.size()) {
-        row_same_bit.erase(row_same_bit.begin() + start, row_same_bit.end());
-    }
 }
 
 void Block::erase(size_t position) {
@@ -278,9 +275,6 @@ void Block::erase_impl(size_t position) {
             }
         }
     }
-    if (position < row_same_bit.size()) {
-        row_same_bit.erase(row_same_bit.begin() + position);
-    }
 }
 
 void Block::erase(const String& name) {
@@ -433,9 +427,6 @@ void Block::set_num_rows(size_t length) {
                 elem.column = elem.column->shrink(length);
             }
         }
-        if (length < row_same_bit.size()) {
-            row_same_bit.resize(length);
-        }
     }
 }
 
@@ -450,9 +441,6 @@ void Block::skip_num_rows(int64_t& length) {
                 elem.column = elem.column->cut(length, origin_rows - length);
             }
         }
-        if (length < row_same_bit.size()) {
-            row_same_bit.assign(row_same_bit.begin() + length, row_same_bit.end());
-        }
     }
 }
 
@@ -783,7 +771,6 @@ DataTypes Block::get_data_types() const {
 void Block::clear() {
     data.clear();
     index_by_name.clear();
-    row_same_bit.clear();
 }
 
 void Block::clear_column_data(int64_t column_size) noexcept {
@@ -802,7 +789,6 @@ void Block::clear_column_data(int64_t column_size) noexcept {
             (*std::move(d.column)).assume_mutable()->clear();
         }
     }
-    row_same_bit.clear();
 }
 
 void Block::erase_tmp_columns() noexcept {
@@ -836,14 +822,12 @@ void Block::swap(Block& other) noexcept {
     SCOPED_SKIP_MEMORY_CHECK();
     data.swap(other.data);
     index_by_name.swap(other.index_by_name);
-    row_same_bit.swap(other.row_same_bit);
 }
 
 void Block::swap(Block&& other) noexcept {
     SCOPED_SKIP_MEMORY_CHECK();
     data = std::move(other.data);
     index_by_name = std::move(other.index_by_name);
-    row_same_bit = std::move(other.row_same_bit);
 }
 
 void Block::shuffle_columns(const std::vector<int>& result_column_ids) {
@@ -1167,9 +1151,6 @@ void MutableBlock::erase(const String& name) {
             ++it;
         }
     }
-    // if (position < row_same_bit.size()) {
-    //     row_same_bit.erase(row_same_bit.begin() + position);
-    // }
 }
 
 Block MutableBlock::to_block(int start_column) {

be/src/vec/core/block.h

@@ -76,7 +76,6 @@ class Block {
     using IndexByName = phmap::flat_hash_map<String, size_t>;
     Container data;
     IndexByName index_by_name;
-    std::vector<bool> row_same_bit;
 
     int64_t _decompress_time_ns = 0;
     int64_t _decompressed_bytes = 0;
@@ -367,21 +366,6 @@ class Block {
     int64_t get_decompressed_bytes() const { return _decompressed_bytes; }
     int64_t get_compress_time() const { return _compress_time_ns; }
 
-    void set_same_bit(std::vector<bool>::const_iterator begin,
-                      std::vector<bool>::const_iterator end) {
-        row_same_bit.insert(row_same_bit.end(), begin, end);
-
-        DCHECK_EQ(row_same_bit.size(), rows());
-    }
-
-    bool get_same_bit(size_t position) {
-        if (position >= row_same_bit.size()) {
-            return false;
-        }
-        return row_same_bit[position];
-    }
-
-    void clear_same_bit() { row_same_bit.clear(); }
 
     // remove tmp columns in block
     // in inverted index apply logic, in order to optimize query performance,

be/src/vec/exec/scan/scanner.cpp

@@ -106,9 +106,6 @@ Status Scanner::get_block(RuntimeState* state, Block* block, bool* eof) {
 
     {
         do {
-            // if step 2 filter all rows of block, and block will be reused to get next rows,
-            // must clear row_same_bit of block, or will get wrong row_same_bit.size() which not equal block.rows()
-            block->clear_same_bit();
             // 1. Get input block from scanner
             {
                 // get block time

be/src/vec/olap/block_reader.cpp

@@ -309,7 +309,7 @@ Status BlockReader::_agg_key_next_block(Block* block, bool* eof) {
             return res;
         }
 
-        if (!_get_next_row_same()) {
+        if (!_next_row.is_same) {
             if (target_block_row == _reader_context.batch_size) {
                 break;
             }
@@ -530,13 +530,5 @@ void BlockReader::_update_agg_value(MutableColumns& columns, int begin, int end,
     }
 }
 
-bool BlockReader::_get_next_row_same() {
-    if (_next_row.is_same) {
-        return true;
-    } else {
-        auto* block = _next_row.block.get();
-        return block->get_same_bit(_next_row.row_pos);
-    }
-}
 #include "common/compile_check_end.h"
 } // namespace doris::vectorized

be/src/vec/olap/block_reader.h

@@ -84,8 +84,6 @@ class BlockReader final : public TabletReader {
 
     void _update_agg_value(MutableColumns& columns, int begin, int end, bool is_close = true);
 
-    bool _get_next_row_same();
-
     // return false if keys of rowsets are mono ascending and disjoint
     bool _rowsets_not_mono_asc_disjoint(const ReaderParams& read_params);