2

Author: xinyiZzz

be/src/olap/iterators.h

@@ -116,7 +116,7 @@ class StorageReadOptions {
     // columns for orderby keys
     std::vector<uint32_t>* read_orderby_key_columns = nullptr;
     IOContext io_ctx;
-    vectorized::VExpr* remaining_vconjunct_root = nullptr;
+    vectorized::VExprContext* remaining_vconjunct_ctx = nullptr;
     // runtime state
     RuntimeState* runtime_state = nullptr;
 };

be/src/olap/reader.cpp

@@ -226,7 +226,7 @@ Status TabletReader::_capture_rs_readers(const ReaderParams& read_params,
     _reader_context.enable_unique_key_merge_on_write = tablet()->enable_unique_key_merge_on_write();
     _reader_context.record_rowids = read_params.record_rowids;
     _reader_context.is_key_column_group = read_params.is_key_column_group;
-    _reader_context.remaining_vconjunct_root = read_params.remaining_vconjunct_root;
+    _reader_context.remaining_vconjunct_ctx = read_params.remaining_vconjunct_ctx;
 
     *valid_rs_readers = *rs_readers;
 

be/src/olap/reader.h

@@ -92,7 +92,7 @@ class TabletReader {
         std::vector<uint32_t>* origin_return_columns = nullptr;
         std::unordered_set<uint32_t>* tablet_columns_convert_to_null_set = nullptr;
         TPushAggOp::type push_down_agg_type_opt = TPushAggOp::NONE;
-        vectorized::VExpr* remaining_vconjunct_root = nullptr;
+        vectorized::VExprContext* remaining_vconjunct_ctx = nullptr;
 
         // used for compaction to record row ids
         bool record_rowids = false;

be/src/olap/rowset/beta_rowset_reader.cpp

@@ -56,7 +56,7 @@ Status BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
     // convert RowsetReaderContext to StorageReadOptions
     _read_options.stats = _stats;
     _read_options.push_down_agg_type_opt = _context->push_down_agg_type_opt;
-    _read_options.remaining_vconjunct_root = _context->remaining_vconjunct_root;
+    _read_options.remaining_vconjunct_ctx = _context->remaining_vconjunct_ctx;
     if (read_context->lower_bound_keys != nullptr) {
         for (int i = 0; i < read_context->lower_bound_keys->size(); ++i) {
             _read_options.key_ranges.emplace_back(&read_context->lower_bound_keys->at(i),

be/src/olap/rowset/rowset_reader_context.h

@@ -58,7 +58,7 @@ struct RowsetReaderContext {
     const DeleteHandler* delete_handler = nullptr;
     OlapReaderStatistics* stats = nullptr;
     RuntimeState* runtime_state = nullptr;
-    vectorized::VExpr* remaining_vconjunct_root = nullptr;
+    vectorized::VExprContext* remaining_vconjunct_ctx = nullptr;
     bool use_page_cache = false;
     int sequence_id_idx = -1;
     int batch_size = 1024;

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

@@ -178,6 +178,12 @@ class SegmentIterator : public RowwiseIterator {
 
     bool _can_evaluated_by_vectorized(ColumnPredicate* predicate);
 
+    Status _extract_remaining_expr_columns(vectorized::VExpr* expr);
+    uint16_t _execute_remaining_expr(uint16_t* sel_rowid_idx, uint16_t selected_size,
+                                     vectorized::Block* block);
+    uint16_t _evaluate_remaining_expr_filter(uint16_t* sel_rowid_idx, uint16_t selected_size,
+                                             const vectorized::IColumn::Filter& filter);
+
     // Dictionary column should do something to initial.
     void _convert_dict_code_for_predicate_if_necessary();
 
@@ -279,18 +285,21 @@ class SegmentIterator : public RowwiseIterator {
     std::vector<ColumnId> _predicate_columns;
     // columns to read after predicate evaluation
     std::vector<ColumnId> _non_predicate_columns;
+    std::vector<ColumnId> _remaining_expr_columns;
     // remember the rowids we've read for the current row block.
     // could be a local variable of next_batch(), kept here to reuse vector memory
     std::vector<rowid_t> _block_rowids;
     bool _is_need_vec_eval = false;
     bool _is_need_short_eval = false;
+    bool _is_need_expr_eval = false;
 
     // fields for vectorization execution
     std::vector<ColumnId>
             _vec_pred_column_ids; // keep columnId of columns for vectorized predicate evaluation
     std::vector<ColumnId>
             _short_cir_pred_column_ids; // keep columnId of columns for short circuit predicate evaluation
     std::vector<bool> _is_pred_column; // columns hold by segmentIter
+    std::vector<bool> _is_remaining_expr_column;
     vectorized::MutableColumns _current_return_columns;
     std::vector<ColumnPredicate*> _pre_eval_block_predicate;
     std::vector<ColumnPredicate*> _short_cir_eval_predicate;
@@ -301,6 +310,8 @@ class SegmentIterator : public RowwiseIterator {
     // second, read non-predicate columns
     // so we need a field to stand for columns first time to read
     std::vector<ColumnId> _first_read_column_ids;
+    std::vector<ColumnId> _second_read_column_ids;
+    std::vector<ColumnId> _columns_to_filter;
     std::vector<int> _schema_block_id_map; // map from schema column id to column idx in Block
 
     // the actual init process is delayed to the first call to next_batch()
@@ -311,6 +322,7 @@ class SegmentIterator : public RowwiseIterator {
     // make a copy of `_opts.column_predicates` in order to make local changes
     std::vector<ColumnPredicate*> _col_predicates;
     std::vector<ColumnPredicate*> _col_preds_except_leafnode_of_andnode;
+    doris::vectorized::VExprContext* _remaining_vconjunct_ctx;
     doris::vectorized::VExpr* _remaining_vconjunct_root;
     std::vector<roaring::Roaring> _pred_except_leafnode_of_andnode_evaluate_result;
     std::unique_ptr<ColumnPredicateInfo> _column_predicate_info;

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

@@ -139,6 +139,11 @@ class RuntimeState {
                _query_options.check_overflow_for_decimal;
     }
 
+    bool enable_remaining_expr_pushdown() const {
+        return _query_options.__isset.enable_remaining_expr_pushdown &&
+               _query_options.enable_remaining_expr_pushdown;
+    }
+
     // Create a codegen object in _codegen. No-op if it has already been called.
     // If codegen is enabled for the query, this is created when the runtime
     // state is created. If codegen is disabled for the query, this is created

be/src/runtime/runtime_state.h

@@ -314,7 +314,7 @@ void Block::check_number_of_rows(bool allow_null_columns) const {
 
 size_t Block::rows() const {
     for (const auto& elem : data) {
-        if (elem.column) {
+        if (elem.column && elem.column->size() != 0) {
             return elem.column->size();
         }
     }

be/src/vec/core/block.cpp

@@ -171,8 +171,9 @@ Status NewOlapScanner::_init_tablet_reader_params(
                 real_parent->_olap_scan_node.push_down_agg_type_opt;
     }
     _tablet_reader_params.version = Version(0, _version);
-    _tablet_reader_params.remaining_vconjunct_root =
-            (_vconjunct_ctx == nullptr) ? nullptr : _vconjunct_ctx->root();
+    if (_state->enable_remaining_expr_pushdown()) {
+        _tablet_reader_params.remaining_vconjunct_ctx = _vconjunct_ctx;
+    }
 
     // Condition
     for (auto& filter : filters) {

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

@@ -80,8 +80,12 @@ Status VScanner::get_block(RuntimeState* state, Block* block, bool* eof) {
 
 Status VScanner::_filter_output_block(Block* block) {
     auto old_rows = block->rows();
-    Status st =
-            VExprContext::filter_block(_vconjunct_ctx, block, _output_tuple_desc->slots().size());
+    Status st;
+    if (_state->enable_remaining_expr_pushdown()) {
+        st = Status::OK();
+    } else {
+        st = VExprContext::filter_block(_vconjunct_ctx, block, _output_tuple_desc->slots().size());
+    }
     _counter.num_rows_unselected += old_rows - block->rows();
     return st;
 }

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

@@ -221,6 +221,8 @@ public class SessionVariable implements Serializable, Writable {
 
     public static final String ENABLE_FUNCTION_PUSHDOWN = "enable_function_pushdown";
 
+    public static final String ENABLE_REMAINING_EXPR_PUSHDOWN = "enable_remaining_expr_pushdown";
+
     public static final String FRAGMENT_TRANSMISSION_COMPRESSION_CODEC = "fragment_transmission_compression_codec";
 
     public static final String ENABLE_LOCAL_EXCHANGE = "enable_local_exchange";
@@ -594,6 +596,9 @@ public class SessionVariable implements Serializable, Writable {
     @VariableMgr.VarAttr(name = ENABLE_FUNCTION_PUSHDOWN)
     public boolean enableFunctionPushdown;
 
+    @VariableMgr.VarAttr(name = ENABLE_REMAINING_EXPR_PUSHDOWN)
+    public boolean enableRemainingExprPushdown = false;
+
     @VariableMgr.VarAttr(name = ENABLE_LOCAL_EXCHANGE)
     public boolean enableLocalExchange = true;
 
@@ -1395,6 +1400,7 @@ public TQueryOptions toThrift() {
         }
 
         tResult.setEnableFunctionPushdown(enableFunctionPushdown);
+        tResult.setEnableRemainingExprPushdown(enableRemainingExprPushdown);
         tResult.setCheckOverflowForDecimal(checkOverflowForDecimal);
         tResult.setFragmentTransmissionCompressionCodec(fragmentTransmissionCompressionCodec);
         tResult.setEnableLocalExchange(enableLocalExchange);

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

@@ -190,6 +190,8 @@ struct TQueryOptions {
   57: optional bool check_overflow_for_decimal = false
 
   58: optional i64 external_sort_bytes_threshold = 0
+
+  59: optional bool enable_remaining_expr_pushdown = false;
 }