[Improvement](scan) Remove redundant predicates on scan node

be/src/pipeline/exec/olap_scan_operator.cpp

@@ -200,9 +200,12 @@ Status OlapScanLocalState::_should_push_down_function_filter(
 }
 
 bool OlapScanLocalState::_should_push_down_common_expr() {
+    return state()->enable_common_expr_pushdown() && _storage_no_merge();
+}
+
+bool OlapScanLocalState::_storage_no_merge() {
     auto& p = _parent->cast<OlapScanOperatorX>();
-    return state()->enable_common_expr_pushdown() &&
-           (p._olap_scan_node.keyType == TKeysType::DUP_KEYS ||
+    return (p._olap_scan_node.keyType == TKeysType::DUP_KEYS ||
             (p._olap_scan_node.keyType == TKeysType::UNIQUE_KEYS &&
              p._olap_scan_node.__isset.enable_unique_key_merge_on_write &&
              p._olap_scan_node.enable_unique_key_merge_on_write));

be/src/pipeline/exec/olap_scan_operator.h

@@ -72,6 +72,8 @@ class OlapScanLocalState final : public ScanLocalState {
 
     bool _should_push_down_common_expr() override;
 
+    bool _storage_no_merge() override;
+
     Status _init_scanners(std::list<vectorized::VScannerSPtr>* scanners) override;
 
     void add_filter_info(int id, const PredicateFilterInfo& info);

be/src/pipeline/exec/scan_operator.cpp

@@ -356,7 +356,7 @@ Status ScanLocalState::_normalize_predicate(const vectorized::VExprSPtr& conjunc
             }
 
             if (pdt == vectorized::VScanNode::PushDownType::ACCEPTABLE &&
-                _is_key_column(slot->col_name())) {
+                (_is_key_column(slot->col_name()) || _storage_no_merge())) {
                 output_expr = nullptr;
                 return Status::OK();
             } else {

be/src/pipeline/exec/scan_operator.h

@@ -59,7 +59,7 @@ class ScanLocalState : public PipelineXLocalState, public vectorized::RuntimeFil
     ENABLE_FACTORY_CREATOR(ScanLocalState);
     ScanLocalState(RuntimeState* state, OperatorXBase* parent);
 
-    virtual Status init(RuntimeState* state, LocalStateInfo& info) override;
+    Status init(RuntimeState* state, LocalStateInfo& info) override;
 
     bool ready_to_read();
 
@@ -89,8 +89,9 @@ class ScanLocalState : public PipelineXLocalState, public vectorized::RuntimeFil
         RETURN_IF_ERROR(_normalize_conjuncts());
         return Status::OK();
     }
-
     virtual bool _should_push_down_common_expr() { return false; }
+
+    virtual bool _storage_no_merge() { return true; }
     virtual bool _is_key_column(const std::string& col_name) { return false; }
     virtual vectorized::VScanNode::PushDownType _should_push_down_bloom_filter() {
         return vectorized::VScanNode::PushDownType::UNACCEPTABLE;

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

@@ -382,14 +382,6 @@ Status NewOlapScanNode::_should_push_down_function_filter(VectorizedFnCall* fn_c
     return Status::OK();
 }
 
-bool NewOlapScanNode::_should_push_down_common_expr() {
-    return _state->enable_common_expr_pushdown() &&
-           (_olap_scan_node.keyType == TKeysType::DUP_KEYS ||
-            (_olap_scan_node.keyType == TKeysType::UNIQUE_KEYS &&
-             _olap_scan_node.__isset.enable_unique_key_merge_on_write &&
-             _olap_scan_node.enable_unique_key_merge_on_write));
-}
-
 // PlanFragmentExecutor will call this method to set scan range
 // Doris scan range is defined in thrift file like this
 // struct TPaloScanRange {
@@ -434,7 +426,7 @@ Status NewOlapScanNode::_init_scanners(std::list<VScannerSPtr>* scanners) {
                 message += conjunct->root()->debug_string();
             }
         }
-        _runtime_profile->add_info_string("RemainedDownPredicates", message);
+        _runtime_profile->add_info_string("RemainedPredicates", message);
     }
 
     if (!_olap_scan_node.output_column_unique_ids.empty()) {

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

@@ -85,7 +85,16 @@ class NewOlapScanNode : public VScanNode {
 
     PushDownType _should_push_down_is_null_predicate() override { return PushDownType::ACCEPTABLE; }
 
-    bool _should_push_down_common_expr() override;
+    bool _should_push_down_common_expr() override {
+        return _state->enable_common_expr_pushdown() && _storage_no_merge();
+    }
+
+    bool _storage_no_merge() override {
+        return (_olap_scan_node.keyType == TKeysType::DUP_KEYS ||
+                (_olap_scan_node.keyType == TKeysType::UNIQUE_KEYS &&
+                 _olap_scan_node.__isset.enable_unique_key_merge_on_write &&
+                 _olap_scan_node.enable_unique_key_merge_on_write));
+    }
 
     Status _init_scanners(std::list<VScannerSPtr>* scanners) override;
 

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

@@ -549,7 +549,8 @@ Status VScanNode::_normalize_predicate(const VExprSPtr& conjunct_expr_root, VExp
                 return Status::OK();
             }
 
-            if (pdt == PushDownType::ACCEPTABLE && _is_key_column(slot->col_name())) {
+            if (pdt == PushDownType::ACCEPTABLE &&
+                (_is_key_column(slot->col_name()) || _storage_no_merge())) {
                 output_expr = nullptr;
                 return Status::OK();
             } else {

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

@@ -228,6 +228,8 @@ class VScanNode : public ExecNode, public RuntimeFilterConsumer {
 
     virtual bool _should_push_down_common_expr() { return false; }
 
+    virtual bool _storage_no_merge() { return false; }
+
     virtual PushDownType _should_push_down_bloom_filter() { return PushDownType::UNACCEPTABLE; }
 
     virtual PushDownType _should_push_down_bitmap_filter() { return PushDownType::UNACCEPTABLE; }