diff --git a/be/src/olap/push_handler.cpp b/be/src/olap/push_handler.cpp
index e494f7db7d7126..783de3e59a036f 100644
--- a/be/src/olap/push_handler.cpp
+++ b/be/src/olap/push_handler.cpp
@@ -646,7 +646,7 @@ Status PushBrokerReader::_get_next_reader() {
 
         init_status = parquet_reader->init_reader(
                 _all_col_names, &_col_name_to_block_idx, _push_down_exprs, _slot_id_to_predicates,
-                _or_predicates, _real_tuple_desc, _default_val_row_desc.get(), _col_name_to_slot_id,
+                _real_tuple_desc, _default_val_row_desc.get(), _col_name_to_slot_id,
                 &_not_single_slot_filter_conjuncts, &_slot_id_to_filter_conjuncts,
                 vectorized::TableSchemaChangeHelper::ConstNode::get_instance(), false);
         _cur_reader = std::move(parquet_reader);
diff --git a/be/src/olap/push_handler.h b/be/src/olap/push_handler.h
index 0b60a81c62465a..fa545f3e848b6a 100644
--- a/be/src/olap/push_handler.h
+++ b/be/src/olap/push_handler.h
@@ -142,7 +142,6 @@ class PushBrokerReader {
     std::unordered_map<std::string, uint32_t> _col_name_to_block_idx;
     vectorized::VExprContextSPtrs _push_down_exprs;
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> _slot_id_to_predicates;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> _or_predicates;
     const std::unordered_map<std::string, int>* _col_name_to_slot_id;
     // single slot filter conjuncts
     std::unordered_map<int, vectorized::VExprContextSPtrs> _slot_id_to_filter_conjuncts;
diff --git a/be/src/pipeline/exec/file_scan_operator.cpp b/be/src/pipeline/exec/file_scan_operator.cpp
index ec3262706678a6..6d5e0d7c24488b 100644
--- a/be/src/pipeline/exec/file_scan_operator.cpp
+++ b/be/src/pipeline/exec/file_scan_operator.cpp
@@ -88,13 +88,14 @@ bool FileScanLocalState::_should_push_down_or_predicate_recursively(
 
 PushDownType FileScanLocalState::_should_push_down_or_predicate(
         const vectorized::VExprContext* expr_ctx) const {
-    auto expr = expr_ctx->root()->get_impl() ? expr_ctx->root()->get_impl() : expr_ctx->root();
-    if (expr->node_type() == TExprNodeType::COMPOUND_PRED &&
-        expr->op() == TExprOpcode::COMPOUND_OR) {
-        if (_should_push_down_or_predicate_recursively(expr)) {
-            return PushDownType::PARTIAL_ACCEPTABLE;
-        }
-    }
+    // TODO(gabriel): Do not push down OR predicate for the time being.
+    //    auto expr = expr_ctx->root()->get_impl() ? expr_ctx->root()->get_impl() : expr_ctx->root();
+    //    if (expr->node_type() == TExprNodeType::COMPOUND_PRED &&
+    //        expr->op() == TExprOpcode::COMPOUND_OR) {
+    //        if (_should_push_down_or_predicate_recursively(expr)) {
+    //            return PushDownType::PARTIAL_ACCEPTABLE;
+    //        }
+    //    }
     return PushDownType::UNACCEPTABLE;
 }
 
diff --git a/be/src/pipeline/exec/scan_operator.cpp b/be/src/pipeline/exec/scan_operator.cpp
index 122ca9dd444160..3da82ed5e0f302 100644
--- a/be/src/pipeline/exec/scan_operator.cpp
+++ b/be/src/pipeline/exec/scan_operator.cpp
@@ -261,8 +261,7 @@ Status ScanLocalState<Derived>::_normalize_conjuncts(RuntimeState* state) {
         auto& conjunct = *it;
         if (conjunct->root()) {
             vectorized::VExprSPtr new_root;
-            RETURN_IF_ERROR(
-                    _normalize_predicate(conjunct.get(), conjunct->root(), new_root, nullptr));
+            RETURN_IF_ERROR(_normalize_predicate(conjunct.get(), conjunct->root(), new_root));
             if (new_root) {
                 conjunct->set_root(new_root);
                 if (_should_push_down_common_expr() &&
@@ -297,132 +296,91 @@ Status ScanLocalState<Derived>::_normalize_conjuncts(RuntimeState* state) {
 template <typename Derived>
 Status ScanLocalState<Derived>::_normalize_predicate(vectorized::VExprContext* context,
                                                      const vectorized::VExprSPtr& root,
-                                                     vectorized::VExprSPtr& output_expr,
-                                                     MutilColumnBlockPredicate* parent) {
+                                                     vectorized::VExprSPtr& output_expr) {
     auto expr_root = root->is_rf_wrapper() ? root->get_impl() : root;
-    if (expr_root->node_type() == TExprNodeType::COMPOUND_PRED &&
-        expr_root->op() == TExprOpcode::COMPOUND_OR) {
-        if (_should_push_down_or_predicate(context) != PushDownType::UNACCEPTABLE) {
-            std::unique_ptr<MutilColumnBlockPredicate> new_root =
-                    OrBlockColumnPredicate::create_unique();
-            DCHECK_GE(expr_root->get_num_children(), 1);
-            for (auto& child : expr_root->children()) {
-                vectorized::VExprSPtr tmp = nullptr;
-                RETURN_IF_ERROR(_normalize_predicate(context, child, tmp, new_root.get()));
-                DCHECK_NE(tmp, nullptr);
-            }
-            if (parent) {
-                parent->add_column_predicate(std::move(new_root));
-            } else {
-                _or_predicates.emplace_back(std::move(new_root));
-            }
-        }
-    } else if (expr_root->node_type() == TExprNodeType::COMPOUND_PRED &&
-               expr_root->op() == TExprOpcode::COMPOUND_AND) {
-        if (!parent) {
-            // AndPredicate is illegal on scan operator unless it is a child of OrPredicate
-            return Status::InternalError(
-                    "And expr must have parent MutilColumnBlockPredicate, but now {}",
-                    expr_root->debug_string());
-        }
-        std::unique_ptr<MutilColumnBlockPredicate> new_root =
-                AndBlockColumnPredicate::create_unique();
-        DCHECK_GE(expr_root->get_num_children(), 1);
-        for (const auto& child : expr_root->children()) {
-            vectorized::VExprSPtr tmp = nullptr;
-            RETURN_IF_ERROR(_normalize_predicate(context, child, tmp, new_root.get()));
-        }
-        DCHECK_GE(new_root->num_of_column_predicate(), 1);
-        parent->add_column_predicate(std::move(new_root));
-    } else {
-        PushDownType pdt = PushDownType::UNACCEPTABLE;
-        if (dynamic_cast<vectorized::VirtualSlotRef*>(expr_root.get())) {
-            // If the expr has virtual slot ref, we need to keep it in the tree.
-            output_expr = expr_root;
-            return Status::OK();
-        }
+    PushDownType pdt = PushDownType::UNACCEPTABLE;
+    if (dynamic_cast<vectorized::VirtualSlotRef*>(expr_root.get())) {
+        // If the expr has virtual slot ref, we need to keep it in the tree.
+        output_expr = expr_root;
+        return Status::OK();
+    }
 
-        SlotDescriptor* slot = nullptr;
-        ColumnValueRangeType* range = nullptr;
-        RETURN_IF_ERROR(_eval_const_conjuncts(context, &pdt));
-        if (pdt == PushDownType::ACCEPTABLE) {
-            output_expr = nullptr;
-            return Status::OK();
-        }
-        std::shared_ptr<vectorized::VSlotRef> slotref;
-        for (const auto& child : expr_root->children()) {
-            if (vectorized::VExpr::expr_without_cast(child)->node_type() !=
-                TExprNodeType::SLOT_REF) {
-                // not a slot ref(column)
-                continue;
-            }
-            slotref = std::dynamic_pointer_cast<vectorized::VSlotRef>(
-                    vectorized::VExpr::expr_without_cast(child));
-        }
-        if (_is_predicate_acting_on_slot(expr_root->children(), &slot, &range)) {
-            Status status = Status::OK();
-            std::visit(
-                    [&](auto& value_range) {
-                        auto r = root;
-                        RETURN_IF_PUSH_DOWN(
-                                _normalize_in_and_eq_predicate(context, r, slot,
-                                                               _slot_id_to_predicates[slot->id()],
-                                                               value_range, &pdt, parent),
-                                status);
-                        RETURN_IF_PUSH_DOWN(
-                                _normalize_not_in_and_not_eq_predicate(
-                                        context, r, slot, _slot_id_to_predicates[slot->id()],
-                                        value_range, &pdt, parent),
-                                status);
-                        RETURN_IF_PUSH_DOWN(
-                                _normalize_is_null_predicate(context, r, slot,
-                                                             _slot_id_to_predicates[slot->id()],
-                                                             value_range, &pdt, parent),
-                                status);
-                        RETURN_IF_PUSH_DOWN(
-                                _normalize_noneq_binary_predicate(
-                                        context, r, slot, _slot_id_to_predicates[slot->id()],
-                                        value_range, &pdt, parent),
-                                status);
-                        RETURN_IF_PUSH_DOWN(
-                                _normalize_bitmap_filter(context, r, slot,
-                                                         _slot_id_to_predicates[slot->id()], &pdt,
-                                                         parent),
-                                status);
-                        RETURN_IF_PUSH_DOWN(
-                                _normalize_bloom_filter(context, r, slot,
-                                                        _slot_id_to_predicates[slot->id()], &pdt,
-                                                        parent),
-                                status);
-                        RETURN_IF_PUSH_DOWN(
-                                _normalize_topn_filter(context, r, slot,
-                                                       _slot_id_to_predicates[slot->id()], &pdt,
-                                                       parent),
-                                status);
-                        if (state()->enable_function_pushdown()) {
-                            RETURN_IF_PUSH_DOWN(_normalize_function_filters(context, slot, &pdt),
-                                                status);
-                        }
-                    },
-                    *range);
-            RETURN_IF_ERROR(status);
-        }
-        if (pdt == PushDownType::ACCEPTABLE && slotref != nullptr &&
-            slotref->data_type()->get_primitive_type() == PrimitiveType::TYPE_VARIANT) {
-            // remaining it in the expr tree, in order to filter by function if the pushdown
-            // predicate is not applied
-            output_expr = expr_root; // remaining in conjunct tree
-            return Status::OK();
+    SlotDescriptor* slot = nullptr;
+    ColumnValueRangeType* range = nullptr;
+    RETURN_IF_ERROR(_eval_const_conjuncts(context, &pdt));
+    if (pdt == PushDownType::ACCEPTABLE) {
+        output_expr = nullptr;
+        return Status::OK();
+    }
+    std::shared_ptr<vectorized::VSlotRef> slotref;
+    for (const auto& child : expr_root->children()) {
+        if (vectorized::VExpr::expr_without_cast(child)->node_type() != TExprNodeType::SLOT_REF) {
+            // not a slot ref(column)
+            continue;
         }
+        slotref = std::dynamic_pointer_cast<vectorized::VSlotRef>(
+                vectorized::VExpr::expr_without_cast(child));
+    }
+    if (_is_predicate_acting_on_slot(expr_root->children(), &slot, &range)) {
+        Status status = Status::OK();
+        std::visit(
+                [&](auto& value_range) {
+                    auto r = root;
+                    RETURN_IF_PUSH_DOWN(
+                            _normalize_in_and_eq_predicate(context, r, slot,
+                                                           _slot_id_to_predicates[slot->id()],
+                                                           value_range, &pdt),
+                            status);
+                    RETURN_IF_PUSH_DOWN(
+                            _normalize_not_in_and_not_eq_predicate(
+                                    context, r, slot, _slot_id_to_predicates[slot->id()],
+                                    value_range, &pdt),
+                            status);
+                    RETURN_IF_PUSH_DOWN(
+                            _normalize_is_null_predicate(context, r, slot,
+                                                         _slot_id_to_predicates[slot->id()],
+                                                         value_range, &pdt),
+                            status);
+                    RETURN_IF_PUSH_DOWN(
+                            _normalize_noneq_binary_predicate(context, r, slot,
+                                                              _slot_id_to_predicates[slot->id()],
+                                                              value_range, &pdt),
+                            status);
+                    RETURN_IF_PUSH_DOWN(
+                            _normalize_bitmap_filter(context, r, slot,
+                                                     _slot_id_to_predicates[slot->id()], &pdt),
+                            status);
+                    RETURN_IF_PUSH_DOWN(
+                            _normalize_bloom_filter(context, r, slot,
+                                                    _slot_id_to_predicates[slot->id()], &pdt),
+                            status);
+                    RETURN_IF_PUSH_DOWN(
+                            _normalize_topn_filter(context, r, slot,
+                                                   _slot_id_to_predicates[slot->id()], &pdt),
+                            status);
+                    if (state()->enable_function_pushdown()) {
+                        RETURN_IF_PUSH_DOWN(_normalize_function_filters(context, slot, &pdt),
+                                            status);
+                    }
+                },
+                *range);
+        RETURN_IF_ERROR(status);
+    }
+    if (pdt == PushDownType::ACCEPTABLE && slotref != nullptr &&
+        slotref->data_type()->get_primitive_type() == PrimitiveType::TYPE_VARIANT) {
+        // remaining it in the expr tree, in order to filter by function if the pushdown
+        // predicate is not applied
+        output_expr = expr_root; // remaining in conjunct tree
+        return Status::OK();
+    }
 
-        if (pdt == PushDownType::ACCEPTABLE && (_is_key_column(slot->col_name()))) {
-            output_expr = nullptr;
-            return Status::OK();
-        } else {
-            // for PARTIAL_ACCEPTABLE and UNACCEPTABLE, do not remove expr from the tree
-            output_expr = root;
-            return Status::OK();
-        }
+    if (pdt == PushDownType::ACCEPTABLE && (_is_key_column(slot->col_name()))) {
+        output_expr = nullptr;
+        return Status::OK();
+    } else {
+        // for PARTIAL_ACCEPTABLE and UNACCEPTABLE, do not remove expr from the tree
+        output_expr = root;
+        return Status::OK();
     }
     output_expr = root;
     return Status::OK();
@@ -431,14 +389,10 @@ Status ScanLocalState<Derived>::_normalize_predicate(vectorized::VExprContext* c
 template <typename Derived>
 Status ScanLocalState<Derived>::_normalize_bloom_filter(
         vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root, SlotDescriptor* slot,
-        std::vector<std::shared_ptr<ColumnPredicate>>& predicates, PushDownType* pdt,
-        MutilColumnBlockPredicate* parent) {
+        std::vector<std::shared_ptr<ColumnPredicate>>& predicates, PushDownType* pdt) {
     std::shared_ptr<ColumnPredicate> pred = nullptr;
     Defer defer = [&]() {
-        if (parent && pred) {
-            DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
-            parent->add_column_predicate(SingleColumnBlockPredicate::create_unique(pred));
-        } else if (pred) {
+        if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
         }
@@ -468,14 +422,10 @@ Status ScanLocalState<Derived>::_normalize_bloom_filter(
 template <typename Derived>
 Status ScanLocalState<Derived>::_normalize_topn_filter(
         vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root, SlotDescriptor* slot,
-        std::vector<std::shared_ptr<ColumnPredicate>>& predicates, PushDownType* pdt,
-        MutilColumnBlockPredicate* parent) {
+        std::vector<std::shared_ptr<ColumnPredicate>>& predicates, PushDownType* pdt) {
     std::shared_ptr<ColumnPredicate> pred = nullptr;
     Defer defer = [&]() {
-        if (parent && pred) {
-            DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
-            parent->add_column_predicate(SingleColumnBlockPredicate::create_unique(pred));
-        } else if (pred) {
+        if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
         }
@@ -498,14 +448,10 @@ Status ScanLocalState<Derived>::_normalize_topn_filter(
 template <typename Derived>
 Status ScanLocalState<Derived>::_normalize_bitmap_filter(
         vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root, SlotDescriptor* slot,
-        std::vector<std::shared_ptr<ColumnPredicate>>& predicates, PushDownType* pdt,
-        MutilColumnBlockPredicate* parent) {
+        std::vector<std::shared_ptr<ColumnPredicate>>& predicates, PushDownType* pdt) {
     std::shared_ptr<ColumnPredicate> pred = nullptr;
     Defer defer = [&]() {
-        if (parent && pred) {
-            DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
-            parent->add_column_predicate(SingleColumnBlockPredicate::create_unique(pred));
-        } else if (pred) {
+        if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
         }
@@ -660,13 +606,10 @@ template <PrimitiveType T>
 Status ScanLocalState<Derived>::_normalize_in_and_eq_predicate(
         vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root, SlotDescriptor* slot,
         std::vector<std::shared_ptr<ColumnPredicate>>& predicates, ColumnValueRange<T>& range,
-        PushDownType* pdt, MutilColumnBlockPredicate* parent) {
+        PushDownType* pdt) {
     std::shared_ptr<ColumnPredicate> pred = nullptr;
     Defer defer = [&]() {
-        if (parent && pred) {
-            DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
-            parent->add_column_predicate(SingleColumnBlockPredicate::create_unique(pred));
-        } else if (pred) {
+        if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
         }
@@ -717,7 +660,7 @@ Status ScanLocalState<Derived>::_normalize_in_and_eq_predicate(
             iter = state->hybrid_set->begin();
         }
 
-        if (iter && !parent) {
+        if (iter) {
             while (iter->has_next()) {
                 // column in (nullptr) is always false so continue to
                 // dispose next item
@@ -758,25 +701,23 @@ Status ScanLocalState<Derived>::_normalize_in_and_eq_predicate(
                             : slot->type(),
                     value, false, _arena);
 
-            if (!parent) {
-                if constexpr (T == TYPE_CHAR || T == TYPE_VARCHAR || T == TYPE_STRING ||
-                              T == TYPE_HLL) {
-                    auto val = StringRef(value.data, value.size);
-                    RETURN_IF_ERROR(_change_value_range<true>(
-                            temp_range, reinterpret_cast<void*>(&val),
-                            ColumnValueRange<T>::add_fixed_value_range, fn_name));
-                } else {
-                    if (sizeof(typename PrimitiveTypeTraits<T>::CppType) != value.size) {
-                        return Status::InternalError(
-                                "PrimitiveType {} meet invalid input value size {}, expect size {}",
-                                T, value.size, sizeof(typename PrimitiveTypeTraits<T>::CppType));
-                    }
-                    RETURN_IF_ERROR(_change_value_range<true>(
-                            temp_range, reinterpret_cast<const void*>(value.data),
-                            ColumnValueRange<T>::add_fixed_value_range, fn_name));
+            if constexpr (T == TYPE_CHAR || T == TYPE_VARCHAR || T == TYPE_STRING ||
+                          T == TYPE_HLL) {
+                auto val = StringRef(value.data, value.size);
+                RETURN_IF_ERROR(_change_value_range<true>(
+                        temp_range, reinterpret_cast<void*>(&val),
+                        ColumnValueRange<T>::add_fixed_value_range, fn_name));
+            } else {
+                if (sizeof(typename PrimitiveTypeTraits<T>::CppType) != value.size) {
+                    return Status::InternalError(
+                            "PrimitiveType {} meet invalid input value size {}, expect size {}", T,
+                            value.size, sizeof(typename PrimitiveTypeTraits<T>::CppType));
                 }
-                range.intersection(temp_range);
+                RETURN_IF_ERROR(_change_value_range<true>(
+                        temp_range, reinterpret_cast<const void*>(value.data),
+                        ColumnValueRange<T>::add_fixed_value_range, fn_name));
             }
+            range.intersection(temp_range);
         } else {
             *pdt = PushDownType::UNACCEPTABLE;
             _eos = true;
@@ -792,13 +733,10 @@ template <PrimitiveType T>
 Status ScanLocalState<Derived>::_normalize_not_in_and_not_eq_predicate(
         vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root, SlotDescriptor* slot,
         std::vector<std::shared_ptr<ColumnPredicate>>& predicates, ColumnValueRange<T>& range,
-        PushDownType* pdt, MutilColumnBlockPredicate* parent) {
+        PushDownType* pdt) {
     std::shared_ptr<ColumnPredicate> pred = nullptr;
     Defer defer = [&]() {
-        if (parent && pred) {
-            DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
-            parent->add_column_predicate(SingleColumnBlockPredicate::create_unique(pred));
-        } else if (pred) {
+        if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
         }
@@ -847,17 +785,15 @@ Status ScanLocalState<Derived>::_normalize_not_in_and_not_eq_predicate(
                 slot->type()->get_primitive_type() == TYPE_VARIANT ? expr->get_child(0)->data_type()
                                                                    : slot->type(),
                 state->hybrid_set, false);
-        if (!parent) {
-            while (iter->has_next()) {
-                // column not in (nullptr) is always true
-                DCHECK(iter->get_value() != nullptr);
-                const auto value = iter->get_value();
-                if (is_fixed_range) {
-                    RETURN_IF_ERROR(_change_value_range<true>(
-                            range, value, ColumnValueRange<T>::remove_fixed_value_range, fn_name));
-                }
-                iter->next();
+        while (iter->has_next()) {
+            // column not in (nullptr) is always true
+            DCHECK(iter->get_value() != nullptr);
+            const auto value = iter->get_value();
+            if (is_fixed_range) {
+                RETURN_IF_ERROR(_change_value_range<true>(
+                        range, value, ColumnValueRange<T>::remove_fixed_value_range, fn_name));
             }
+            iter->next();
         }
     } else if (TExprNodeType::BINARY_PRED == expr->node_type()) {
         DCHECK(expr->get_num_children() == 2);
@@ -883,22 +819,20 @@ Status ScanLocalState<Derived>::_normalize_not_in_and_not_eq_predicate(
                             ? expr->get_child(0)->data_type()
                             : slot->type(),
                     value, false, _arena);
-            if (!parent) {
-                auto fn_name = std::string("");
-                if constexpr (T == TYPE_CHAR || T == TYPE_VARCHAR || T == TYPE_STRING ||
-                              T == TYPE_HLL) {
-                    auto val = StringRef(value.data, value.size);
-                    if (is_fixed_range) {
-                        RETURN_IF_ERROR(_change_value_range<true>(
-                                range, reinterpret_cast<void*>(&val),
-                                ColumnValueRange<T>::remove_fixed_value_range, fn_name));
-                    }
-                } else {
-                    if (is_fixed_range) {
-                        RETURN_IF_ERROR(_change_value_range<true>(
-                                range, reinterpret_cast<const void*>(value.data),
-                                ColumnValueRange<T>::remove_fixed_value_range, fn_name));
-                    }
+            auto fn_name = std::string("");
+            if constexpr (T == TYPE_CHAR || T == TYPE_VARCHAR || T == TYPE_STRING ||
+                          T == TYPE_HLL) {
+                auto val = StringRef(value.data, value.size);
+                if (is_fixed_range) {
+                    RETURN_IF_ERROR(_change_value_range<true>(
+                            range, reinterpret_cast<void*>(&val),
+                            ColumnValueRange<T>::remove_fixed_value_range, fn_name));
+                }
+            } else {
+                if (is_fixed_range) {
+                    RETURN_IF_ERROR(_change_value_range<true>(
+                            range, reinterpret_cast<const void*>(value.data),
+                            ColumnValueRange<T>::remove_fixed_value_range, fn_name));
                 }
             }
         } else {
@@ -986,13 +920,10 @@ template <PrimitiveType T>
 Status ScanLocalState<Derived>::_normalize_is_null_predicate(
         vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root, SlotDescriptor* slot,
         std::vector<std::shared_ptr<ColumnPredicate>>& predicates, ColumnValueRange<T>& range,
-        PushDownType* pdt, MutilColumnBlockPredicate* parent) {
+        PushDownType* pdt) {
     std::shared_ptr<ColumnPredicate> pred = nullptr;
     Defer defer = [&]() {
-        if (parent && pred) {
-            DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
-            parent->add_column_predicate(SingleColumnBlockPredicate::create_unique(pred));
-        } else if (pred) {
+        if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
         }
@@ -1013,22 +944,18 @@ Status ScanLocalState<Derived>::_normalize_is_null_predicate(
         pred = NullPredicate::create_shared(
                 _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(), true,
                 T);
-        if (!parent) {
-            auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(
-                    slot->is_nullable(), range.precision(), range.scale());
-            temp_range.set_contain_null(true);
-            range.intersection(temp_range);
-        }
+        auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(
+                slot->is_nullable(), range.precision(), range.scale());
+        temp_range.set_contain_null(true);
+        range.intersection(temp_range);
     } else if (fn_call->fn().name.function_name == "is_not_null_pred") {
         pred = NullPredicate::create_shared(
                 _parent->intermediate_row_desc().get_column_id(slot->id()), slot->col_name(), false,
                 T);
-        if (!parent) {
-            auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(
-                    slot->is_nullable(), range.precision(), range.scale());
-            temp_range.set_contain_null(false);
-            range.intersection(temp_range);
-        }
+        auto temp_range = ColumnValueRange<T>::create_empty_column_value_range(
+                slot->is_nullable(), range.precision(), range.scale());
+        temp_range.set_contain_null(false);
+        range.intersection(temp_range);
     }
     return Status::OK();
 }
@@ -1038,13 +965,10 @@ template <PrimitiveType T>
 Status ScanLocalState<Derived>::_normalize_noneq_binary_predicate(
         vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root, SlotDescriptor* slot,
         std::vector<std::shared_ptr<ColumnPredicate>>& predicates, ColumnValueRange<T>& range,
-        PushDownType* pdt, MutilColumnBlockPredicate* parent) {
+        PushDownType* pdt) {
     std::shared_ptr<ColumnPredicate> pred = nullptr;
     Defer defer = [&]() {
-        if (parent && pred) {
-            DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
-            parent->add_column_predicate(SingleColumnBlockPredicate::create_unique(pred));
-        } else if (pred) {
+        if (pred) {
             DCHECK(*pdt != PushDownType::UNACCEPTABLE) << root->debug_string();
             predicates.emplace_back(pred);
         }
@@ -1101,18 +1025,16 @@ Status ScanLocalState<Derived>::_normalize_noneq_binary_predicate(
                 throw Exception(
                         Status::InternalError("Unsupported function name: {}", function_name));
             }
-            if (!parent) {
-                if constexpr (T == TYPE_CHAR || T == TYPE_VARCHAR || T == TYPE_STRING ||
-                              T == TYPE_HLL) {
-                    auto val = StringRef(value.data, value.size);
-                    RETURN_IF_ERROR(_change_value_range<false>(range, reinterpret_cast<void*>(&val),
-                                                               ColumnValueRange<T>::add_value_range,
-                                                               function_name));
-                } else {
-                    RETURN_IF_ERROR(_change_value_range<false>(
-                            range, reinterpret_cast<const void*>(value.data),
-                            ColumnValueRange<T>::add_value_range, function_name));
-                }
+            if constexpr (T == TYPE_CHAR || T == TYPE_VARCHAR || T == TYPE_STRING ||
+                          T == TYPE_HLL) {
+                auto val = StringRef(value.data, value.size);
+                RETURN_IF_ERROR(_change_value_range<false>(range, reinterpret_cast<void*>(&val),
+                                                           ColumnValueRange<T>::add_value_range,
+                                                           function_name));
+            } else {
+                RETURN_IF_ERROR(_change_value_range<false>(
+                        range, reinterpret_cast<const void*>(value.data),
+                        ColumnValueRange<T>::add_value_range, function_name));
             }
         } else {
             *pdt = PushDownType::UNACCEPTABLE;
diff --git a/be/src/pipeline/exec/scan_operator.h b/be/src/pipeline/exec/scan_operator.h
index f1f1624e423220..03f59b9c1f5a9b 100644
--- a/be/src/pipeline/exec/scan_operator.h
+++ b/be/src/pipeline/exec/scan_operator.h
@@ -250,23 +250,22 @@ class ScanLocalState : public ScanLocalStateBase {
     // Normalize a conjunct and try to convert it to column predicate recursively.
     Status _normalize_predicate(vectorized::VExprContext* context,
                                 const vectorized::VExprSPtr& root,
-                                vectorized::VExprSPtr& output_expr,
-                                MutilColumnBlockPredicate* parent);
+                                vectorized::VExprSPtr& output_expr);
     Status _eval_const_conjuncts(vectorized::VExprContext* expr_ctx, PushDownType* pdt);
 
     Status _normalize_bloom_filter(vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root,
                                    SlotDescriptor* slot,
                                    std::vector<std::shared_ptr<ColumnPredicate>>& predicates,
-                                   PushDownType* pdt, MutilColumnBlockPredicate* parent);
+                                   PushDownType* pdt);
     Status _normalize_topn_filter(vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root,
                                   SlotDescriptor* slot,
                                   std::vector<std::shared_ptr<ColumnPredicate>>& predicates,
-                                  PushDownType* pdt, MutilColumnBlockPredicate* parent);
+                                  PushDownType* pdt);
 
     Status _normalize_bitmap_filter(vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root,
                                     SlotDescriptor* slot,
                                     std::vector<std::shared_ptr<ColumnPredicate>>& predicates,
-                                    PushDownType* pdt, MutilColumnBlockPredicate* parent);
+                                    PushDownType* pdt);
 
     Status _normalize_function_filters(vectorized::VExprContext* expr_ctx, SlotDescriptor* slot,
                                        PushDownType* pdt);
@@ -278,25 +277,23 @@ class ScanLocalState : public ScanLocalStateBase {
     Status _normalize_in_and_eq_predicate(vectorized::VExprContext* expr_ctx,
                                           vectorized::VExprSPtr& root, SlotDescriptor* slot,
                                           std::vector<std::shared_ptr<ColumnPredicate>>& predicates,
-                                          ColumnValueRange<T>& range, PushDownType* pdt,
-                                          MutilColumnBlockPredicate* parent);
+                                          ColumnValueRange<T>& range, PushDownType* pdt);
     template <PrimitiveType T>
     Status _normalize_not_in_and_not_eq_predicate(
             vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root, SlotDescriptor* slot,
             std::vector<std::shared_ptr<ColumnPredicate>>& predicates, ColumnValueRange<T>& range,
-            PushDownType* pdt, MutilColumnBlockPredicate* parent);
+            PushDownType* pdt);
 
     template <PrimitiveType T>
     Status _normalize_noneq_binary_predicate(
             vectorized::VExprContext* expr_ctx, vectorized::VExprSPtr& root, SlotDescriptor* slot,
             std::vector<std::shared_ptr<ColumnPredicate>>& predicates, ColumnValueRange<T>& range,
-            PushDownType* pdt, MutilColumnBlockPredicate* parent);
+            PushDownType* pdt);
     template <PrimitiveType T>
     Status _normalize_is_null_predicate(vectorized::VExprContext* expr_ctx,
                                         vectorized::VExprSPtr& root, SlotDescriptor* slot,
                                         std::vector<std::shared_ptr<ColumnPredicate>>& predicates,
-                                        ColumnValueRange<T>& range, PushDownType* pdt,
-                                        MutilColumnBlockPredicate* parent);
+                                        ColumnValueRange<T>& range, PushDownType* pdt);
 
     template <bool IsFixed, PrimitiveType PrimitiveType, typename ChangeFixedValueRangeFunc>
     Status _change_value_range(ColumnValueRange<PrimitiveType>& range, const void* value,
diff --git a/be/src/vec/exec/format/parquet/vparquet_group_reader.h b/be/src/vec/exec/format/parquet/vparquet_group_reader.h
index 0ea390c9bff86b..0cf2b36eb1b6bd 100644
--- a/be/src/vec/exec/format/parquet/vparquet_group_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_group_reader.h
@@ -93,7 +93,6 @@ class RowGroupReader : public ProfileCollector {
 
         phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>
                 slot_id_to_predicates;
-        std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
         bool can_lazy_read = false;
         // block->rows() returns the number of rows of the first column,
         // so we should check and resize the first column
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
index 21a49673f0837f..10621f0ca5f5e2 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
@@ -329,7 +329,6 @@ Status ParquetReader::init_reader(
         const VExprContextSPtrs& conjuncts,
         phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
                 slot_id_to_predicates,
-        std::vector<std::shared_ptr<MutilColumnBlockPredicate>>& or_predicates,
         const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
         const std::unordered_map<std::string, int>* colname_to_slot_id,
         const VExprContextSPtrs* not_single_slot_filter_conjuncts,
@@ -384,7 +383,6 @@ Status ParquetReader::init_reader(
     // build column predicates for column lazy read
     _lazy_read_ctx.conjuncts = conjuncts;
     _lazy_read_ctx.slot_id_to_predicates = slot_id_to_predicates;
-    _lazy_read_ctx.or_predicates = or_predicates;
     return Status::OK();
 }
 
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.h b/be/src/vec/exec/format/parquet/vparquet_reader.h
index d6f16d595b8931..53f0c425d29403 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.h
@@ -121,7 +121,6 @@ class ParquetReader : public GenericReader {
             const VExprContextSPtrs& conjuncts,
             phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
                     slot_id_to_predicates,
-            std::vector<std::shared_ptr<MutilColumnBlockPredicate>>& or_predicates,
             const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
             const std::unordered_map<std::string, int>* colname_to_slot_id,
             const VExprContextSPtrs* not_single_slot_filter_conjuncts,
diff --git a/be/src/vec/exec/format/table/hive_reader.cpp b/be/src/vec/exec/format/table/hive_reader.cpp
index 1af8479668357e..f0465d4c4c0b87 100644
--- a/be/src/vec/exec/format/table/hive_reader.cpp
+++ b/be/src/vec/exec/format/table/hive_reader.cpp
@@ -216,7 +216,6 @@ Status HiveParquetReader::init_reader(
         const VExprContextSPtrs& conjuncts,
         phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
                 slot_id_to_predicates,
-        std::vector<std::shared_ptr<MutilColumnBlockPredicate>>& or_predicates,
         const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
         const std::unordered_map<std::string, int>* colname_to_slot_id,
         const VExprContextSPtrs* not_single_slot_filter_conjuncts,
@@ -289,9 +288,8 @@ Status HiveParquetReader::init_reader(
 
     return parquet_reader->init_reader(
             read_table_col_names, col_name_to_block_idx, conjuncts, slot_id_to_predicates,
-            or_predicates, tuple_descriptor, row_descriptor, colname_to_slot_id,
-            not_single_slot_filter_conjuncts, slot_id_to_filter_conjuncts, table_info_node_ptr,
-            true, column_ids, filter_column_ids);
+            tuple_descriptor, row_descriptor, colname_to_slot_id, not_single_slot_filter_conjuncts,
+            slot_id_to_filter_conjuncts, table_info_node_ptr, true, column_ids, filter_column_ids);
 }
 
 ColumnIdResult HiveParquetReader::_create_column_ids(const FieldDescriptor* field_desc,
diff --git a/be/src/vec/exec/format/table/hive_reader.h b/be/src/vec/exec/format/table/hive_reader.h
index 8a3385b1d626c8..469260011c56f5 100644
--- a/be/src/vec/exec/format/table/hive_reader.h
+++ b/be/src/vec/exec/format/table/hive_reader.h
@@ -91,7 +91,6 @@ class HiveParquetReader final : public HiveReader {
             const VExprContextSPtrs& conjuncts,
             phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
                     slot_id_to_predicates,
-            std::vector<std::shared_ptr<MutilColumnBlockPredicate>>& or_predicates,
             const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
             const std::unordered_map<std::string, int>* colname_to_slot_id,
             const VExprContextSPtrs* not_single_slot_filter_conjuncts,
diff --git a/be/src/vec/exec/format/table/hudi_reader.cpp b/be/src/vec/exec/format/table/hudi_reader.cpp
index 9a2b708d72af9a..c9c0497d2074fe 100644
--- a/be/src/vec/exec/format/table/hudi_reader.cpp
+++ b/be/src/vec/exec/format/table/hudi_reader.cpp
@@ -36,7 +36,6 @@ Status HudiParquetReader::init_reader(
         const VExprContextSPtrs& conjuncts,
         phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
                 slot_id_to_predicates,
-        std::vector<std::shared_ptr<MutilColumnBlockPredicate>>& or_predicates,
         const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
         const std::unordered_map<std::string, int>* colname_to_slot_id,
         const VExprContextSPtrs* not_single_slot_filter_conjuncts,
@@ -50,10 +49,10 @@ Status HudiParquetReader::init_reader(
     RETURN_IF_ERROR(gen_table_info_node_by_field_id(
             _params, _range.table_format_params.hudi_params.schema_id, tuple_descriptor,
             *field_desc));
-    return parquet_reader->init_reader(
-            read_table_col_names, col_name_to_block_idx, conjuncts, slot_id_to_predicates,
-            or_predicates, tuple_descriptor, row_descriptor, colname_to_slot_id,
-            not_single_slot_filter_conjuncts, slot_id_to_filter_conjuncts, table_info_node_ptr);
+    return parquet_reader->init_reader(read_table_col_names, col_name_to_block_idx, conjuncts,
+                                       slot_id_to_predicates, tuple_descriptor, row_descriptor,
+                                       colname_to_slot_id, not_single_slot_filter_conjuncts,
+                                       slot_id_to_filter_conjuncts, table_info_node_ptr);
 }
 
 #include "common/compile_check_end.h"
diff --git a/be/src/vec/exec/format/table/hudi_reader.h b/be/src/vec/exec/format/table/hudi_reader.h
index 08d9035612b055..f7dd120200042c 100644
--- a/be/src/vec/exec/format/table/hudi_reader.h
+++ b/be/src/vec/exec/format/table/hudi_reader.h
@@ -54,7 +54,6 @@ class HudiParquetReader final : public HudiReader {
             const VExprContextSPtrs& conjuncts,
             phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
                     slot_id_to_predicates,
-            std::vector<std::shared_ptr<MutilColumnBlockPredicate>>& or_predicates,
             const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
             const std::unordered_map<std::string, int>* colname_to_slot_id,
             const VExprContextSPtrs* not_single_slot_filter_conjuncts,
diff --git a/be/src/vec/exec/format/table/iceberg_reader.cpp b/be/src/vec/exec/format/table/iceberg_reader.cpp
index 8b9800a7ab3dde..8e15f7f6e4b13b 100644
--- a/be/src/vec/exec/format/table/iceberg_reader.cpp
+++ b/be/src/vec/exec/format/table/iceberg_reader.cpp
@@ -178,10 +178,9 @@ Status IcebergTableReader::_equality_delete_base(
         }
         if (auto* parquet_reader = typeid_cast<ParquetReader*>(delete_reader.get())) {
             phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-            std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
             RETURN_IF_ERROR(parquet_reader->init_reader(
-                    equality_delete_col_names, &delete_col_name_to_block_idx, {}, tmp,
-                    or_predicates, nullptr, nullptr, nullptr, nullptr, nullptr,
+                    equality_delete_col_names, &delete_col_name_to_block_idx, {}, tmp, nullptr,
+                    nullptr, nullptr, nullptr, nullptr,
                     TableSchemaChangeHelper::ConstNode::get_instance(), false));
         } else if (auto* orc_reader = typeid_cast<OrcReader*>(delete_reader.get())) {
             RETURN_IF_ERROR(orc_reader->init_reader(&equality_delete_col_names,
@@ -448,7 +447,6 @@ Status IcebergParquetReader::init_reader(
         const VExprContextSPtrs& conjuncts,
         phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
                 slot_id_to_predicates,
-        std::vector<std::shared_ptr<MutilColumnBlockPredicate>>& or_predicates,
         const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
         const std::unordered_map<std::string, int>* colname_to_slot_id,
         const VExprContextSPtrs* not_single_slot_filter_conjuncts,
@@ -493,9 +491,8 @@ Status IcebergParquetReader::init_reader(
     }
     return parquet_reader->init_reader(
             _all_required_col_names, _col_name_to_block_idx, conjuncts, slot_id_to_predicates,
-            or_predicates, tuple_descriptor, row_descriptor, colname_to_slot_id,
-            not_single_slot_filter_conjuncts, slot_id_to_filter_conjuncts, table_info_node_ptr,
-            true, column_ids, filter_column_ids);
+            tuple_descriptor, row_descriptor, colname_to_slot_id, not_single_slot_filter_conjuncts,
+            slot_id_to_filter_conjuncts, table_info_node_ptr, true, column_ids, filter_column_ids);
 }
 
 ColumnIdResult IcebergParquetReader::_create_column_ids(const FieldDescriptor* field_desc,
@@ -566,11 +563,10 @@ Status IcebergParquetReader ::_read_position_delete_file(const TFileRangeDesc* d
                                         READ_DELETE_FILE_BATCH_SIZE, &_state->timezone_obj(),
                                         _io_ctx, _state, _meta_cache);
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
     RETURN_IF_ERROR(parquet_delete_reader.init_reader(
             delete_file_col_names,
             const_cast<std::unordered_map<std::string, uint32_t>*>(&DELETE_COL_NAME_TO_BLOCK_IDX),
-            {}, tmp, or_predicates, nullptr, nullptr, nullptr, nullptr, nullptr,
+            {}, tmp, nullptr, nullptr, nullptr, nullptr, nullptr,
             TableSchemaChangeHelper::ConstNode::get_instance(), false));
 
     std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
diff --git a/be/src/vec/exec/format/table/iceberg_reader.h b/be/src/vec/exec/format/table/iceberg_reader.h
index 29f41ef714a1a7..995b78915b5279 100644
--- a/be/src/vec/exec/format/table/iceberg_reader.h
+++ b/be/src/vec/exec/format/table/iceberg_reader.h
@@ -176,7 +176,6 @@ class IcebergParquetReader final : public IcebergTableReader {
             const VExprContextSPtrs& conjuncts,
             phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
                     slot_id_to_predicates,
-            std::vector<std::shared_ptr<MutilColumnBlockPredicate>>& or_predicates,
             const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
             const std::unordered_map<std::string, int>* colname_to_slot_id,
             const VExprContextSPtrs* not_single_slot_filter_conjuncts,
diff --git a/be/src/vec/exec/format/table/paimon_reader.h b/be/src/vec/exec/format/table/paimon_reader.h
index 734c98c20edabd..30cd788ce89163 100644
--- a/be/src/vec/exec/format/table/paimon_reader.h
+++ b/be/src/vec/exec/format/table/paimon_reader.h
@@ -107,7 +107,6 @@ class PaimonParquetReader final : public PaimonReader {
             const VExprContextSPtrs& conjuncts,
             phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>>&
                     slot_id_to_predicates,
-            std::vector<std::shared_ptr<MutilColumnBlockPredicate>>& or_predicates,
             const TupleDescriptor* tuple_descriptor, const RowDescriptor* row_descriptor,
             const std::unordered_map<std::string, int>* colname_to_slot_id,
             const VExprContextSPtrs* not_single_slot_filter_conjuncts,
@@ -122,10 +121,10 @@ class PaimonParquetReader final : public PaimonReader {
                 _params, _range.table_format_params.paimon_params.schema_id, tuple_descriptor,
                 *field_desc));
 
-        return parquet_reader->init_reader(
-                read_table_col_names, col_name_to_block_idx, conjuncts, slot_id_to_predicates,
-                or_predicates, tuple_descriptor, row_descriptor, colname_to_slot_id,
-                not_single_slot_filter_conjuncts, slot_id_to_filter_conjuncts, table_info_node_ptr);
+        return parquet_reader->init_reader(read_table_col_names, col_name_to_block_idx, conjuncts,
+                                           slot_id_to_predicates, tuple_descriptor, row_descriptor,
+                                           colname_to_slot_id, not_single_slot_filter_conjuncts,
+                                           slot_id_to_filter_conjuncts, table_info_node_ptr);
     }
 };
 #include "common/compile_check_end.h"
diff --git a/be/src/vec/exec/scan/file_scanner.cpp b/be/src/vec/exec/scan/file_scanner.cpp
index 414620b0cbe15b..8da6c762137a8f 100644
--- a/be/src/vec/exec/scan/file_scanner.cpp
+++ b/be/src/vec/exec/scan/file_scanner.cpp
@@ -1218,9 +1218,6 @@ Status FileScanner::_init_parquet_reader(std::unique_ptr<ParquetReader>&& parque
             _local_state
                     ? _local_state->cast<pipeline::FileScanLocalState>()._slot_id_to_predicates
                     : phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> {};
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates =
-            _local_state ? _local_state->cast<pipeline::FileScanLocalState>()._or_predicates
-                         : std::vector<std::shared_ptr<MutilColumnBlockPredicate>> {};
     if (range.__isset.table_format_params &&
         range.table_format_params.table_format_type == "iceberg") {
         std::unique_ptr<IcebergParquetReader> iceberg_reader = IcebergParquetReader::create_unique(
@@ -1228,7 +1225,7 @@ Status FileScanner::_init_parquet_reader(std::unique_ptr<ParquetReader>&& parque
                 _io_ctx.get(), file_meta_cache_ptr);
         init_status = iceberg_reader->init_reader(
                 _file_col_names, &_src_block_name_to_idx, _push_down_conjuncts,
-                slot_id_to_predicates, or_predicates, _real_tuple_desc, _default_val_row_desc.get(),
+                slot_id_to_predicates, _real_tuple_desc, _default_val_row_desc.get(),
                 _col_name_to_slot_id, &_not_single_slot_filter_conjuncts,
                 &_slot_id_to_filter_conjuncts);
         _cur_reader = std::move(iceberg_reader);
@@ -1239,7 +1236,7 @@ Status FileScanner::_init_parquet_reader(std::unique_ptr<ParquetReader>&& parque
                 file_meta_cache_ptr);
         init_status = paimon_reader->init_reader(
                 _file_col_names, &_src_block_name_to_idx, _push_down_conjuncts,
-                slot_id_to_predicates, or_predicates, _real_tuple_desc, _default_val_row_desc.get(),
+                slot_id_to_predicates, _real_tuple_desc, _default_val_row_desc.get(),
                 _col_name_to_slot_id, &_not_single_slot_filter_conjuncts,
                 &_slot_id_to_filter_conjuncts);
         RETURN_IF_ERROR(paimon_reader->init_row_filters());
@@ -1251,7 +1248,7 @@ Status FileScanner::_init_parquet_reader(std::unique_ptr<ParquetReader>&& parque
                 file_meta_cache_ptr);
         init_status = hudi_reader->init_reader(
                 _file_col_names, &_src_block_name_to_idx, _push_down_conjuncts,
-                slot_id_to_predicates, or_predicates, _real_tuple_desc, _default_val_row_desc.get(),
+                slot_id_to_predicates, _real_tuple_desc, _default_val_row_desc.get(),
                 _col_name_to_slot_id, &_not_single_slot_filter_conjuncts,
                 &_slot_id_to_filter_conjuncts);
         _cur_reader = std::move(hudi_reader);
@@ -1261,7 +1258,7 @@ Status FileScanner::_init_parquet_reader(std::unique_ptr<ParquetReader>&& parque
                                                             &_is_file_slot, file_meta_cache_ptr);
         init_status = hive_reader->init_reader(
                 _file_col_names, &_src_block_name_to_idx, _push_down_conjuncts,
-                slot_id_to_predicates, or_predicates, _real_tuple_desc, _default_val_row_desc.get(),
+                slot_id_to_predicates, _real_tuple_desc, _default_val_row_desc.get(),
                 _col_name_to_slot_id, &_not_single_slot_filter_conjuncts,
                 &_slot_id_to_filter_conjuncts);
         _cur_reader = std::move(hive_reader);
@@ -1278,7 +1275,7 @@ Status FileScanner::_init_parquet_reader(std::unique_ptr<ParquetReader>&& parque
                 _real_tuple_desc, *parquet_meta, tvf_info_node));
         init_status = parquet_reader->init_reader(
                 _file_col_names, &_src_block_name_to_idx, _push_down_conjuncts,
-                slot_id_to_predicates, or_predicates, _real_tuple_desc, _default_val_row_desc.get(),
+                slot_id_to_predicates, _real_tuple_desc, _default_val_row_desc.get(),
                 _col_name_to_slot_id, &_not_single_slot_filter_conjuncts,
                 &_slot_id_to_filter_conjuncts, tvf_info_node);
         _cur_reader = std::move(parquet_reader);
@@ -1309,7 +1306,7 @@ Status FileScanner::_init_parquet_reader(std::unique_ptr<ParquetReader>&& parque
 
         init_status = parquet_reader->init_reader(
                 _file_col_names, &_src_block_name_to_idx, _push_down_conjuncts,
-                slot_id_to_predicates, or_predicates, _real_tuple_desc, _default_val_row_desc.get(),
+                slot_id_to_predicates, _real_tuple_desc, _default_val_row_desc.get(),
                 _col_name_to_slot_id, &_not_single_slot_filter_conjuncts,
                 &_slot_id_to_filter_conjuncts, load_info_node);
         _cur_reader = std::move(parquet_reader);
diff --git a/be/test/pipeline/operator/scan_normalize_predicate_test.cpp b/be/test/pipeline/operator/scan_normalize_predicate_test.cpp
index 081444ed02ad65..ea290e02ecd3b5 100644
--- a/be/test/pipeline/operator/scan_normalize_predicate_test.cpp
+++ b/be/test/pipeline/operator/scan_normalize_predicate_test.cpp
@@ -56,7 +56,7 @@ TEST_F(ScanNormalizePredicate, test1) {
     auto conjunct_expr_root =
             MockSlotRef::create_mock_context(0, std::make_shared<DataTypeInt64>());
     auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                conjunct_expr_root->root(), new_root, nullptr);
+                                                conjunct_expr_root->root(), new_root);
     EXPECT_TRUE(st) << st.msg();
     std::cout << new_root->debug_string() << std::endl;
 }
@@ -85,7 +85,7 @@ TEST_F(ScanNormalizePredicate, test_eval_const_conjuncts1) {
     vectorized::VExprSPtr new_root;
     auto conjunct_expr_root = ctx;
     auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                conjunct_expr_root->root(), new_root, nullptr);
+                                                conjunct_expr_root->root(), new_root);
 
     EXPECT_EQ(new_root, nullptr);
     EXPECT_TRUE(local_state->_scan_dependency->ready());
@@ -115,7 +115,7 @@ TEST_F(ScanNormalizePredicate, test_eval_const_conjuncts2) {
     vectorized::VExprSPtr new_root;
     auto conjunct_expr_root = ctx;
     auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                conjunct_expr_root->root(), new_root, nullptr);
+                                                conjunct_expr_root->root(), new_root);
     EXPECT_EQ(new_root, nullptr);
     EXPECT_TRUE(local_state->_scan_dependency->ready());
     EXPECT_TRUE(local_state->_eos);
@@ -163,7 +163,7 @@ TEST_F(ScanNormalizePredicate, test_eval_const_conjuncts4) {
     vectorized::VExprSPtr new_root;
     auto conjunct_expr_root = ctx;
     auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                conjunct_expr_root->root(), new_root, nullptr);
+                                                conjunct_expr_root->root(), new_root);
     EXPECT_TRUE(st.ok());
     std::cout << st.msg() << std::endl;
 }
@@ -205,7 +205,7 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot1) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         std::cout << st.msg() << std::endl;
     }
@@ -264,7 +264,7 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot2) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         std::cout << st.msg() << std::endl;
     }
@@ -288,7 +288,7 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot2) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         std::cout << st.msg() << std::endl;
     }
@@ -349,8 +349,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot3) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
     EXPECT_TRUE(local_state->_scan_dependency->ready());
     EXPECT_TRUE(local_state->_eos);
@@ -388,8 +388,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot4) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 
     EXPECT_TRUE(local_state->_slot_id_to_value_range.contains(SlotId));
@@ -446,8 +446,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot5) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 
     EXPECT_TRUE(local_state->_slot_id_to_value_range.contains(SlotId));
@@ -510,8 +510,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot6) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 
     EXPECT_TRUE(local_state->_slot_id_to_value_range.contains(SlotId));
@@ -574,8 +574,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot7) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 
     EXPECT_TRUE(local_state->_scan_dependency->ready());
@@ -620,8 +620,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot8) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 
     auto& output_range = local_state->_slot_id_to_value_range[SlotId];
@@ -673,8 +673,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot10) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 }
 
@@ -715,8 +715,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot11) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 }
 
@@ -760,8 +760,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot12) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 
     auto& output_range = local_state->_slot_id_to_value_range[SlotId];
@@ -817,8 +817,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot13) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 
     auto& output_range = local_state->_slot_id_to_value_range[SlotId];
@@ -874,8 +874,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot14) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 
     auto& output_range = local_state->_slot_id_to_value_range[SlotId];
@@ -935,8 +935,8 @@ TEST_F(ScanNormalizePredicate, test_is_predicate_acting_on_slot15) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
     }
 
     auto& output_range = local_state->_slot_id_to_value_range[SlotId];
@@ -1002,7 +1002,7 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
@@ -1044,7 +1044,7 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
         EXPECT_TRUE(local_state->_slot_id_to_value_range.contains(SlotId));
@@ -1086,8 +1086,8 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
         EXPECT_EQ(new_root, nullptr);
         EXPECT_TRUE(local_state->_slot_id_to_value_range.contains(SlotId));
     }
@@ -1114,7 +1114,7 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
         EXPECT_TRUE(local_state->_slot_id_to_value_range.contains(SlotId));
@@ -1139,8 +1139,8 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
         auto& output_range = local_state->_slot_id_to_value_range[SlotId];
         std::visit(
                 [](auto&& arg) {
@@ -1178,8 +1178,8 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
         auto& output_range = local_state->_slot_id_to_value_range[SlotId];
         std::visit(
                 [](auto&& arg) {
@@ -1221,7 +1221,7 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
@@ -1281,7 +1281,7 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
@@ -1338,7 +1338,7 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
@@ -1398,7 +1398,7 @@ TEST_F(ScanNormalizePredicate, test_double_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
@@ -1480,7 +1480,7 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
@@ -1523,7 +1523,7 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
         EXPECT_TRUE(local_state->_slot_id_to_value_range.contains(SlotId));
@@ -1567,8 +1567,8 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
         EXPECT_EQ(new_root, nullptr);
         EXPECT_TRUE(local_state->_slot_id_to_value_range.contains(SlotId));
     }
@@ -1596,7 +1596,7 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
         EXPECT_TRUE(local_state->_slot_id_to_value_range.contains(SlotId));
@@ -1625,8 +1625,8 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
         auto& output_range = local_state->_slot_id_to_value_range[SlotId];
         std::visit(
                 [](auto&& arg) {
@@ -1665,8 +1665,8 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
 
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
-        EXPECT_TRUE(local_state->_normalize_predicate(
-                conjunct_expr_root.get(), conjunct_expr_root->root(), new_root, nullptr));
+        EXPECT_TRUE(local_state->_normalize_predicate(conjunct_expr_root.get(),
+                                                      conjunct_expr_root->root(), new_root));
         auto& output_range = local_state->_slot_id_to_value_range[SlotId];
         std::visit(
                 [](auto&& arg) {
@@ -1710,7 +1710,7 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
@@ -1771,7 +1771,7 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
@@ -1829,7 +1829,7 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
@@ -1890,7 +1890,7 @@ TEST_F(ScanNormalizePredicate, test_timestamptz_predicate) {
         vectorized::VExprSPtr new_root;
         auto conjunct_expr_root = ctx;
         auto st = local_state->_normalize_predicate(conjunct_expr_root.get(),
-                                                    conjunct_expr_root->root(), new_root, nullptr);
+                                                    conjunct_expr_root->root(), new_root);
         EXPECT_TRUE(st.ok());
         EXPECT_EQ(new_root, nullptr);
 
diff --git a/be/test/vec/exec/format/parquet/parquet_expr_test.cpp b/be/test/vec/exec/format/parquet/parquet_expr_test.cpp
index ebc949f706cb95..40f3891df4667d 100644
--- a/be/test/vec/exec/format/parquet/parquet_expr_test.cpp
+++ b/be/test/vec/exec/format/parquet/parquet_expr_test.cpp
@@ -282,10 +282,9 @@ class ParquetExprTest : public testing::Test {
         p_reader->set_file_reader(local_file_reader);
         colname_to_slot_id.emplace("int64_col", 2);
         phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-        std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
         static_cast<void>(p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp,
-                                                or_predicates, tuple_desc, nullptr,
-                                                &colname_to_slot_id, nullptr, nullptr));
+                                                tuple_desc, nullptr, &colname_to_slot_id, nullptr,
+                                                nullptr));
 
         size_t meta_size;
         static_cast<void>(parse_thrift_footer(p_reader->_file_reader, &doris_file_metadata,
diff --git a/be/test/vec/exec/format/parquet/parquet_read_lines.cpp b/be/test/vec/exec/format/parquet/parquet_read_lines.cpp
index 745fed9ad7fa31..df5d1edb63ba8a 100644
--- a/be/test/vec/exec/format/parquet/parquet_read_lines.cpp
+++ b/be/test/vec/exec/format/parquet/parquet_read_lines.cpp
@@ -153,10 +153,8 @@ static void read_parquet_lines(std::vector<std::string> numeric_types,
 
     std::unordered_map<std::string, ColumnValueRangeType> colname_to_value_range;
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
-    static_cast<void>(p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp,
-                                            or_predicates, nullptr, nullptr, nullptr, nullptr,
-                                            nullptr));
+    static_cast<void>(p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, nullptr,
+                                            nullptr, nullptr, nullptr, nullptr));
     std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
             partition_columns;
     std::unordered_map<std::string, VExprContextSPtr> missing_columns;
diff --git a/be/test/vec/exec/format/parquet/parquet_reader_test.cpp b/be/test/vec/exec/format/parquet/parquet_reader_test.cpp
index 351068f4bf8357..f458f0455c2392 100644
--- a/be/test/vec/exec/format/parquet/parquet_reader_test.cpp
+++ b/be/test/vec/exec/format/parquet/parquet_reader_test.cpp
@@ -152,10 +152,8 @@ TEST_F(ParquetReaderTest, normal) {
     runtime_state.set_desc_tbl(desc_tbl);
 
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
-    static_cast<void>(p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp,
-                                            or_predicates, nullptr, nullptr, nullptr, nullptr,
-                                            nullptr));
+    static_cast<void>(p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, nullptr,
+                                            nullptr, nullptr, nullptr, nullptr));
     std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
             partition_columns;
     std::unordered_map<std::string, VExprContextSPtr> missing_columns;
@@ -219,9 +217,8 @@ TEST_F(ParquetReaderTest, uuid_varbinary) {
     runtime_state.set_desc_tbl(desc_tbl);
 
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
-    st = p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, or_predicates,
-                               nullptr, nullptr, nullptr, nullptr, nullptr);
+    st = p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, nullptr, nullptr,
+                               nullptr, nullptr, nullptr);
     EXPECT_TRUE(st.ok()) << st;
     std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
             partition_columns;
@@ -294,9 +291,8 @@ TEST_F(ParquetReaderTest, varbinary_varbinary) {
     runtime_state.set_desc_tbl(desc_tbl);
 
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
-    st = p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, or_predicates,
-                               nullptr, nullptr, nullptr, nullptr, nullptr);
+    st = p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, nullptr, nullptr,
+                               nullptr, nullptr, nullptr);
     EXPECT_TRUE(st.ok()) << st;
     std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
             partition_columns;
@@ -371,9 +367,8 @@ TEST_F(ParquetReaderTest, varbinary_string) {
     runtime_state.set_desc_tbl(desc_tbl);
 
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
-    st = p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, or_predicates,
-                               nullptr, nullptr, nullptr, nullptr, nullptr);
+    st = p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, nullptr, nullptr,
+                               nullptr, nullptr, nullptr);
     EXPECT_TRUE(st.ok()) << st;
     std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
             partition_columns;
@@ -448,9 +443,8 @@ TEST_F(ParquetReaderTest, varbinary_string2) {
     runtime_state.set_desc_tbl(desc_tbl);
 
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
-    st = p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, or_predicates,
-                               nullptr, nullptr, nullptr, nullptr, nullptr);
+    st = p_reader->init_reader(column_names, &col_name_to_block_idx, {}, tmp, nullptr, nullptr,
+                               nullptr, nullptr, nullptr);
     EXPECT_TRUE(st.ok()) << st;
     std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
             partition_columns;
diff --git a/be/test/vec/exec/format/table/hive/hive_reader_test.cpp b/be/test/vec/exec/format/table/hive/hive_reader_test.cpp
index 421f4218742fed..224844710a4710 100644
--- a/be/test/vec/exec/format/table/hive/hive_reader_test.cpp
+++ b/be/test/vec/exec/format/table/hive/hive_reader_test.cpp
@@ -573,11 +573,9 @@ TEST_F(HiveReaderTest, read_hive_parquet_file) {
     const std::unordered_map<int, VExprContextSPtrs>* slot_id_to_filter_conjuncts = nullptr;
 
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
     st = hive_reader->init_reader(table_col_names, &col_name_to_block_idx, conjuncts, tmp,
-                                  or_predicates, tuple_descriptor, row_descriptor,
-                                  colname_to_slot_id, not_single_slot_filter_conjuncts,
-                                  slot_id_to_filter_conjuncts);
+                                  tuple_descriptor, row_descriptor, colname_to_slot_id,
+                                  not_single_slot_filter_conjuncts, slot_id_to_filter_conjuncts);
     ASSERT_TRUE(st.ok()) << st;
 
     std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
diff --git a/be/test/vec/exec/format/table/iceberg/iceberg_reader_test.cpp b/be/test/vec/exec/format/table/iceberg/iceberg_reader_test.cpp
index 40373f05ad93e6..6587ff8d3a19ea 100644
--- a/be/test/vec/exec/format/table/iceberg/iceberg_reader_test.cpp
+++ b/be/test/vec/exec/format/table/iceberg/iceberg_reader_test.cpp
@@ -573,11 +573,9 @@ TEST_F(IcebergReaderTest, read_iceberg_parquet_file) {
     const std::unordered_map<int, VExprContextSPtrs>* slot_id_to_filter_conjuncts = nullptr;
 
     phmap::flat_hash_map<int, std::vector<std::shared_ptr<ColumnPredicate>>> tmp;
-    std::vector<std::shared_ptr<MutilColumnBlockPredicate>> or_predicates;
     st = iceberg_reader->init_reader(table_col_names, &col_name_to_block_idx, conjuncts, tmp,
-                                     or_predicates, tuple_descriptor, row_descriptor,
-                                     colname_to_slot_id, not_single_slot_filter_conjuncts,
-                                     slot_id_to_filter_conjuncts);
+                                     tuple_descriptor, row_descriptor, colname_to_slot_id,
+                                     not_single_slot_filter_conjuncts, slot_id_to_filter_conjuncts);
     ASSERT_TRUE(st.ok()) << st;
 
     std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>