[feature](scan) Implement BE-side value predicate pushdown for MOR tables

Enable value column predicates to be pushed down to storage layer for
MOR (Merge-On-Read) tables when controlled by the session variable
enable_mor_value_predicate_pushdown_tables. This allows inverted indexes
on value columns to be utilized for filtering, improving query performance
on dedup-only/insert-only MOR workloads.

Key changes:
- Propagate enable_mor_value_predicate_pushdown flag from thrift scan node
  through OlapScanner -> ReaderParams -> RowsetReaderContext -> BetaRowsetReader
- Extend _should_push_down_value_predicates() to push value predicates for
  all rowsets when the flag is set
- Skip __DORIS_DELETE_SIGN__ predicate during per-segment pushdown to prevent
  delete markers from being filtered before merge
- Revert scan_operator.cpp to only remove VExpr from conjuncts for key columns,
  preserving VExpr as post-merge safety net for value columns
- Add regression tests covering dedup, delete-sign, and delete-predicate scenarios

Author: dataroaring

be/src/olap/rowset/beta_rowset_reader.cpp

@@ -356,7 +356,8 @@ bool BetaRowsetReader::_should_push_down_value_predicates() const {
            (((_rowset->start_version() == 0 || _rowset->start_version() == 2) &&
              !_rowset->_rowset_meta->is_segments_overlapping() &&
              _read_context->sequence_id_idx == -1) ||
-            _read_context->enable_unique_key_merge_on_write);
+            _read_context->enable_unique_key_merge_on_write ||
+            _read_context->enable_mor_value_predicate_pushdown);
 }
 #include "common/compile_check_end.h"
 } // namespace doris

be/src/olap/rowset/rowset_reader_context.h

@@ -100,6 +100,9 @@ struct RowsetReaderContext {
     std::shared_ptr<segment_v2::AnnTopNRuntime> ann_topn_runtime;
 
     uint64_t condition_cache_digest = 0;
+
+    // When true, push down value predicates for MOR tables
+    bool enable_mor_value_predicate_pushdown = false;
 };
 
 } // namespace doris

be/src/olap/tablet_reader.cpp

@@ -208,6 +208,8 @@ Status TabletReader::_capture_rs_readers(const ReaderParams& read_params) {
     _reader_context.merged_rows = &_merged_rows;
     _reader_context.delete_bitmap = read_params.delete_bitmap;
     _reader_context.enable_unique_key_merge_on_write = tablet()->enable_unique_key_merge_on_write();
+    _reader_context.enable_mor_value_predicate_pushdown =
+            read_params.enable_mor_value_predicate_pushdown;
     _reader_context.record_rowids = read_params.record_rowids;
     _reader_context.rowid_conversion = read_params.rowid_conversion;
     _reader_context.is_key_column_group = read_params.is_key_column_group;
@@ -515,9 +517,17 @@ Status TabletReader::_init_conditions_param(const ReaderParams& read_params) {
         }
     }
 
+    int32_t delete_sign_idx = _tablet_schema->delete_sign_idx();
     for (auto predicate : predicates) {
         auto column = _tablet_schema->column(predicate->column_id());
         if (column.aggregation() != FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE) {
+            // When MOR value predicate pushdown is enabled, drop __DORIS_DELETE_SIGN__
+            // from storage-layer predicates entirely. Delete sign must only be evaluated
+            // post-merge via VExpr to prevent deleted rows from reappearing.
+            if (_reader_context.enable_mor_value_predicate_pushdown && delete_sign_idx >= 0 &&
+                predicate->column_id() == static_cast<uint32_t>(delete_sign_idx)) {
+                continue;
+            }
             _value_col_predicates.push_back(predicate);
         } else {
             _col_predicates.push_back(predicate);

be/src/olap/tablet_reader.h

@@ -191,6 +191,9 @@ class TabletReader {
 
         bool is_segcompaction = false;
 
+        // Enable value predicate pushdown for MOR tables
+        bool enable_mor_value_predicate_pushdown = false;
+
         std::vector<RowwiseIteratorUPtr>* segment_iters_ptr = nullptr;
 
         void check_validation() const;

be/src/pipeline/exec/scan_operator.cpp

@@ -439,8 +439,7 @@ Status ScanLocalState<Derived>::_normalize_predicate(vectorized::VExprContext* c
         return Status::OK();
     }
 
-    if (pdt == PushDownType::ACCEPTABLE &&
-        (_is_key_column(slot->col_name()) || _should_push_down_mor_value_predicate())) {
+    if (pdt == PushDownType::ACCEPTABLE && _is_key_column(slot->col_name())) {
         output_expr = nullptr;
         return Status::OK();
     } else {

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

@@ -480,6 +480,13 @@ Status OlapScanner::_init_tablet_reader_params(
     if (!_state->skip_storage_engine_merge()) {
         auto* olap_scan_local_state = (pipeline::OlapScanLocalState*)_local_state;
         TOlapScanNode& olap_scan_node = olap_scan_local_state->olap_scan_node();
+
+        // Set MOR value predicate pushdown flag
+        if (olap_scan_node.__isset.enable_mor_value_predicate_pushdown &&
+            olap_scan_node.enable_mor_value_predicate_pushdown) {
+            _tablet_reader_params.enable_mor_value_predicate_pushdown = true;
+        }
+
         // order by table keys optimization for topn
         // will only read head/tail of data file since it's already sorted by keys
         if (olap_scan_node.__isset.sort_info && !olap_scan_node.sort_info.is_asc_order.empty()) {

regression-test/data/data_model_p0/unique/test_mor_value_predicate_pushdown.out

@@ -1,29 +1,69 @@
 -- This file is automatically generated. You should know what you did if you want to edit this
 -- !select_disabled --
+2	200	world
 3	300	test
 
 -- !select_enabled_tablename --
+2	200	world
 3	300	test
 
 -- !select_enabled_fullname --
+2	200	world
 3	300	test
 
 -- !select_enabled_wildcard --
+2	200	world
 3	300	test
 
--- !select_deleted_row --
+-- !select_eq_predicate --
+2	200	world
 
 -- !select_not_in_list --
+2	200	world
 3	300	test
 
--- !select_latest_version --
-1	200	second
+-- !select_dedup_all --
+1	100	first
 2	300	third
+3	500	fifth
 
--- !select_old_version --
+-- !select_dedup_eq --
+2	300	third
+
+-- !select_dedup_none --
+
+-- !select_delete_range --
+3	300	test
+
+-- !select_delete_eq --
+
+-- !select_delete_all --
+1	100	hello
+3	300	test
+
+-- !select_delpred_range --
+3	300	test
+
+-- !select_delpred_eq --
+
+-- !select_delpred_all --
+1	100	hello
+3	300	test
+
+-- !select_update_disabled_old --
+
+-- !select_update_disabled_new --
+1	500	new
+
+-- !select_update_enabled_old --
+1	100	old
+
+-- !select_update_enabled_new --
+1	500	new
 
--- !select_new_version --
-1	200	second
+-- !select_update_enabled_range --
+1	500	new
+3	300	keep
 
 -- !select_multiple_tables --
 2	200