2

Author: xinyiZzz

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

@@ -1287,6 +1287,7 @@ void SegmentIterator::_vec_init_lazy_materialization() {
             std::set<ColumnId> non_pred_set(_non_predicate_columns.begin(),
                                             _non_predicate_columns.end());
 
+            DCHECK(_second_read_column_ids.empty());
             // _second_read_column_ids must be empty. Otherwise _lazy_materialization_read must not false.
             for (int i = 0; i < _schema.num_column_ids(); i++) {
                 auto cid = _schema.column_id(i);
@@ -1704,15 +1705,19 @@ Status SegmentIterator::next_batch(vectorized::Block* block) {
                     }
 
                     // block->rows() takes the size of the first column by default. If the first column is no predicate column,
-                    // it has not been read yet. Specify the column that has been read to calculate rows().
-                    block->set_effective_col(_schema_block_id_map[*_common_expr_columns.begin()]);
+                    // it has not been read yet. add a const column that has been read to calculate rows().
+                    if (block->rows() == 0) {
+                        auto res_column = vectorized::ColumnString::create();
+                        res_column->insert_data("", 0);
+                        auto col_const = vectorized::ColumnConst::create(std::move(res_column),
+                                                                         selected_size);
+                        block->replace_by_position(0, std::move(col_const));
+                    }
                     DCHECK(block->columns() > _schema_block_id_map[*_common_expr_columns.begin()]);
                     _output_index_result_column(sel_rowid_idx, selected_size, block);
 
                     block->shrink_char_type_column_suffix_zero(_char_type_idx);
                     RETURN_IF_ERROR(_execute_common_expr(sel_rowid_idx, selected_size, block));
-
-                    block->set_effective_col(INT_MIN);
                 }
             } else if (_is_need_expr_eval) {
                 for (auto cid : _second_read_column_ids) {
@@ -1731,12 +1736,16 @@ Status SegmentIterator::next_batch(vectorized::Block* block) {
                 sel_rowid_idx[i] = i;
             }
 
-            block->set_effective_col(_schema_block_id_map[*_common_expr_columns.begin()]);
+            if (block->rows() == 0) {
+                auto res_column = vectorized::ColumnString::create();
+                res_column->insert_data("", 0);
+                auto col_const =
+                        vectorized::ColumnConst::create(std::move(res_column), selected_size);
+                block->replace_by_position(0, std::move(col_const));
+            }
             _output_index_result_column(nullptr, 0, block);
             block->shrink_char_type_column_suffix_zero(_char_type_idx);
             RETURN_IF_ERROR(_execute_common_expr(sel_rowid_idx, selected_size, block));
-
-            block->set_effective_col(INT_MIN);
         }
 
         if (UNLIKELY(_opts.record_rowids)) {

be/src/vec/core/block.cpp

@@ -198,7 +198,6 @@ void Block::erase(size_t position) {
     DCHECK(position < data.size()) << fmt::format(
             "Position out of bound in Block::erase(), max position = {}", data.size() - 1);
 
-    DCHECK(position != _effective_col);
     erase_impl(position);
 }
 
@@ -329,9 +328,6 @@ void Block::check_number_of_rows(bool allow_null_columns) const {
 }
 
 size_t Block::rows() const {
-    if (_effective_col != INT_MIN) {
-        return data[_effective_col].column->size();
-    }
     for (const auto& elem : data) {
         if (elem.column) {
             return elem.column->size();
@@ -628,7 +624,6 @@ void Block::clear() {
     data.clear();
     index_by_name.clear();
     row_same_bit.clear();
-    _effective_col = INT_MIN;
 }
 
 void Block::clear_column_data(int column_size) noexcept {

be/src/vec/core/block.h

@@ -75,7 +75,6 @@ class Block {
     IndexByName index_by_name;
     std::vector<bool> row_same_bit;
 
-    int _effective_col = INT_MIN;
     int64_t _decompress_time_ns = 0;
     int64_t _decompressed_bytes = 0;
 
@@ -203,9 +202,6 @@ class Block {
 
     std::string each_col_size() const;
 
-    // Specify column index to calculate rows(), default rows() uses the size of the first column.
-    void set_effective_col(int col) { _effective_col = col; }
-
     // Cut the rows in block, use in LIMIT operation
     void set_num_rows(size_t length);
 

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

@@ -443,9 +443,9 @@ Status NewOlapScanNode::_init_scanners(std::list<VScanner*>* scanners) {
             // add scanner to pool before doing prepare.
             // so that scanner can be automatically deconstructed if prepare failed.
             _scanner_pool.add(scanner);
-            RETURN_IF_ERROR(scanner->prepare(*scan_range, scanner_ranges, _vconjunct_ctx_ptr.get(),
-                                             _olap_filters, _filter_predicates,
-                                             _push_down_functions, _common_vexpr_ctxs_pushdown));
+            RETURN_IF_ERROR(scanner->prepare(
+                    *scan_range, scanner_ranges, _vconjunct_ctx_ptr.get(), _olap_filters,
+                    _filter_predicates, _push_down_functions, _common_vexpr_ctxs_pushdown.get()));
             scanners->push_back((VScanner*)scanner);
             disk_set.insert(scanner->scan_disk());
         }

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

@@ -52,10 +52,10 @@ Status NewOlapScanner::prepare(const TPaloScanRange& scan_range,
                                const std::vector<TCondition>& filters,
                                const FilterPredicates& filter_predicates,
                                const std::vector<FunctionFilter>& function_filters,
-                               VExprContext* common_vexpr_ctxs_pushdown) {
+                               VExprContext** common_vexpr_ctxs_pushdown) {
     if (common_vexpr_ctxs_pushdown != nullptr) {
         // Copy common_vexpr_ctxs_pushdown from scan node to this scanner's _common_vexpr_ctxs_pushdown, just necessary.
-        RETURN_IF_ERROR(common_vexpr_ctxs_pushdown->clone(_state, &_common_vexpr_ctxs_pushdown));
+        RETURN_IF_ERROR((*common_vexpr_ctxs_pushdown)->clone(_state, &_common_vexpr_ctxs_pushdown));
     }
 
     // set limit to reduce end of rowset and segment mem use

be/src/vec/exec/scan/new_olap_scanner.h

@@ -45,7 +45,7 @@ class NewOlapScanner : public VScanner {
                    VExprContext** vconjunct_ctx_ptr, const std::vector<TCondition>& filters,
                    const FilterPredicates& filter_predicates,
                    const std::vector<FunctionFilter>& function_filters,
-                   VExprContext* common_vexpr_ctxs_pushdown);
+                   VExprContext** common_vexpr_ctxs_pushdown);
 
     const std::string& scan_disk() const { return _tablet->data_dir()->path(); }
 

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

@@ -389,7 +389,7 @@ void VScanNode::release_resource(RuntimeState* state) {
         (*ctx)->close(state);
     }
     if (_common_vexpr_ctxs_pushdown) {
-        _common_vexpr_ctxs_pushdown->close(state);
+        (*_common_vexpr_ctxs_pushdown)->close(state);
     }
     _scanner_pool.clear();
 
@@ -461,7 +461,7 @@ Status VScanNode::_normalize_conjuncts() {
             if (new_root) {
                 (*_vconjunct_ctx_ptr)->set_root(new_root);
                 if (_should_push_down_common_expr()) {
-                    _common_vexpr_ctxs_pushdown = *_vconjunct_ctx_ptr;
+                    _common_vexpr_ctxs_pushdown = std::move(_vconjunct_ctx_ptr);
                     _vconjunct_ctx_ptr.reset(nullptr);
                 }
             } else { // All conjucts are pushed down as predicate column

be/src/vec/exec/scan/vscan_node.h

@@ -261,7 +261,7 @@ class VScanNode : public ExecNode {
     // Every time vconjunct_ctx_ptr is updated, the old ctx will be stored in this vector
     // so that it will be destroyed uniformly at the end of the query.
     std::vector<std::unique_ptr<VExprContext*>> _stale_vexpr_ctxs;
-    VExprContext* _common_vexpr_ctxs_pushdown = nullptr;
+    std::unique_ptr<VExprContext*> _common_vexpr_ctxs_pushdown = nullptr;
 
     // If sort info is set, push limit to each scanner;
     int64_t _limit_per_scanner = -1;

fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java

@@ -769,6 +769,7 @@ public class SessionVariable implements Serializable, Writable {
     public void initFuzzyModeVariables() {
         Random random = new Random(System.currentTimeMillis());
         this.parallelExecInstanceNum = random.nextInt(8) + 1;
+        this.enableCommonExprPushdown = random.nextBoolean();
         this.enableLocalExchange = random.nextBoolean();
         // This will cause be dead loop, disable it first
         // this.disableJoinReorder = random.nextBoolean();