diff --git a/be/src/olap/delete_handler.cpp b/be/src/olap/delete_handler.cpp
index 06bd08777b7556..b35f91abce72b4 100644
--- a/be/src/olap/delete_handler.cpp
+++ b/be/src/olap/delete_handler.cpp
@@ -27,6 +27,7 @@
 #include <string>
 #include <vector>
 
+#include "common/logging.h"
 #include "gen_cpp/olap_file.pb.h"
 #include "olap/olap_common.h"
 #include "olap/olap_cond.h"
@@ -224,11 +225,25 @@ bool DeleteHandler::_parse_condition(const std::string& condition_str, TConditio
     return true;
 }
 
+void DeleteHandler::_merge_del_conds() {
+    _merged_del_conds.filter_version = _version;
+    _merged_del_conds.del_cond = new(std::nothrow) Conditions();
+    CHECK(_merged_del_conds.del_cond != nullptr) << "fail to malloc Conditions. size=" << sizeof(Conditions);
+    _merged_del_conds.del_cond->set_tablet_schema(_schema);
+
+    for (const auto& del_cond : _del_conds) {
+        DCHECK_LE(del_cond.filter_version, _version);
+        _merged_del_conds.del_cond->merge_condition(del_cond.del_cond->sorted_conds());
+    }
+}
+
 OLAPStatus DeleteHandler::init(const TabletSchema& schema,
                                const DelPredicateArray& delete_conditions, int64_t version) {
     DCHECK(!_is_inited) << "reinitialize delete handler.";
     DCHECK(version >= 0) << "invalid parameters. version=" << version;
 
+    _version = version;
+    _schema = &schema;
     for (const auto& delete_condition : delete_conditions) {
         // 跳过版本号大于version的过滤条件
         if (delete_condition.version() > version) {
@@ -279,6 +294,19 @@ OLAPStatus DeleteHandler::init(const TabletSchema& schema,
         _del_conds.push_back(temp);
     }
 
+    for (auto& del_cond : _del_conds) {
+        del_cond.del_cond->normalize();
+    }
+
+    // Do lower cost evaluation at first.
+    std::sort(_del_conds.begin(), _del_conds.end(),
+              [] (const DeleteConditions& left,
+                  const DeleteConditions& right) {
+                  return left.del_cond->eval_cost() < right.del_cond->eval_cost();
+              });
+
+    // _merge_del_conds();
+
     _is_inited = true;
 
     return OLAP_SUCCESS;
@@ -309,11 +337,13 @@ void DeleteHandler::finalize() {
         return;
     }
 
+    delete _merged_del_conds.del_cond;
     for (auto& cond : _del_conds) {
         cond.del_cond->finalize();
         delete cond.del_cond;
     }
     _del_conds.clear();
+
     _is_inited = false;
 }
 
diff --git a/be/src/olap/delete_handler.h b/be/src/olap/delete_handler.h
index aea828dd12a674..3a46fa943d7d60 100644
--- a/be/src/olap/delete_handler.h
+++ b/be/src/olap/delete_handler.h
@@ -134,9 +134,14 @@ class DeleteHandler {
     // Use regular expression to extract 'column_name', 'op' and 'operands'
     bool _parse_condition(const std::string& condition_str, TCondition* condition);
 
+    void _merge_del_conds();
+
     bool _is_inited = false;
+    int64_t _version = 0;
+    const TabletSchema* _schema = nullptr;
     // DeleteConditions in _del_conds are in 'OR' relationship
     std::vector<DeleteConditions> _del_conds;
+    DeleteConditions _merged_del_conds;
 
     DISALLOW_COPY_AND_ASSIGN(DeleteHandler);
 };
diff --git a/be/src/olap/olap_cond.cpp b/be/src/olap/olap_cond.cpp
index 4508e2e4dffbd5..60e2272ff901d9 100644
--- a/be/src/olap/olap_cond.cpp
+++ b/be/src/olap/olap_cond.cpp
@@ -87,15 +87,6 @@ static CondOp parse_op_type(const string& op) {
     return OP_NULL;
 }
 
-Cond::~Cond() {
-    delete operand_field;
-    for (auto& it : operand_set) {
-        delete it;
-    }
-    min_value_field = nullptr;
-    max_value_field = nullptr;
-}
-
 OLAPStatus Cond::init(const TCondition& tcond, const TabletColumn& column) {
     // Parse op type
     op = parse_op_type(tcond.condition_op);
@@ -108,7 +99,7 @@ OLAPStatus Cond::init(const TCondition& tcond, const TabletColumn& column) {
         // 'is null' or 'is not null'
         DCHECK_EQ(tcond.condition_values.size(), 1);
         auto operand = tcond.condition_values.begin();
-        std::unique_ptr<WrapperField> f(WrapperField::create(column, operand->length()));
+        std::shared_ptr<WrapperField> f(WrapperField::create(column, operand->length()));
         if (f == nullptr) {
             OLAP_LOG_WARNING("Create field failed. [name=%s, operand=%s, op_type=%d]",
                              tcond.column_name.c_str(), operand->c_str(), op);
@@ -119,11 +110,11 @@ OLAPStatus Cond::init(const TCondition& tcond, const TabletColumn& column) {
         } else {
             f->set_not_null();
         }
-        operand_field = f.release();
+        operand_field = f;
     } else if (op != OP_IN && op != OP_NOT_IN) {
         DCHECK_EQ(tcond.condition_values.size(), 1);
         auto operand = tcond.condition_values.begin();
-        std::unique_ptr<WrapperField> f(WrapperField::create(column, operand->length()));
+        std::shared_ptr<WrapperField> f(WrapperField::create(column, operand->length()));
         if (f == nullptr) {
             OLAP_LOG_WARNING("Create field failed. [name=%s, operand=%s, op_type=%d]",
                              tcond.column_name.c_str(), operand->c_str(), op);
@@ -135,12 +126,12 @@ OLAPStatus Cond::init(const TCondition& tcond, const TabletColumn& column) {
                              tcond.column_name.c_str(), operand->c_str(), op);
             return res;
         }
-        operand_field = f.release();
+        operand_field = f;
     } else {
         DCHECK(op == OP_IN || op == OP_NOT_IN);
         DCHECK(!tcond.condition_values.empty());
         for (auto& operand : tcond.condition_values) {
-            std::unique_ptr<WrapperField> f(WrapperField::create(column, operand.length()));
+            std::shared_ptr<WrapperField> f(WrapperField::create(column, operand.length()));
             if (f == nullptr) {
                 OLAP_LOG_WARNING("Create field failed. [name=%s, operand=%s, op_type=%d]",
                                  tcond.column_name.c_str(), operand.c_str(), op);
@@ -152,21 +143,18 @@ OLAPStatus Cond::init(const TCondition& tcond, const TabletColumn& column) {
                                  tcond.column_name.c_str(), operand.c_str(), op);
                 return res;
             }
-            if (min_value_field == nullptr || f->cmp(min_value_field) < 0) {
-                min_value_field = f.get();
+            if (min_value_field == nullptr || f->cmp(min_value_field.get()) < 0) {
+                min_value_field = f;
             }
 
-            if (max_value_field == nullptr || f->cmp(max_value_field) > 0) {
-                max_value_field = f.get();
+            if (max_value_field == nullptr || f->cmp(max_value_field.get()) > 0) {
+                max_value_field = f;
             }
 
-            auto insert_result = operand_set.insert(f.get());
+            auto insert_result = operand_set.insert(f);
             if (!insert_result.second) {
                 LOG(WARNING) << "Duplicate operand in in-predicate.[condition=" << operand << "]";
-                // Duplicated, let std::unique_ptr delete field
-            } else {
-                // Normal case, release this std::unique_ptr
-                f.release();
+                // Duplicated, let unique_ptr delete field
             }
         }
     }
@@ -174,6 +162,383 @@ OLAPStatus Cond::init(const TCondition& tcond, const TabletColumn& column) {
     return OLAP_SUCCESS;
 }
 
+OLAPStatus Cond::intersection_cond(const Cond& other) {
+    DCHECK(op == other.op ||
+          (op == OP_NOT_IN && other.op == OP_NE) ||
+          (op == OP_NE && other.op == OP_NOT_IN) ||
+          (op == OP_IN && other.op == OP_EQ) ||
+          (op == OP_EQ && other.op == OP_IN) ||
+          (op == OP_LT && other.op == OP_LE) ||
+          (op == OP_LE && other.op == OP_LT) ||
+          (op == OP_GT && other.op == OP_GE) ||
+          (op == OP_GE && other.op == OP_GT)) << "op: " << op << ", other.op: " << other.op;
+    switch (op) {
+        case OP_EQ:
+            if (other.op == OP_EQ) {
+                if (operand_field->field()->compare_cell(*operand_field, *(other.operand_field)) != 0) {
+                    // No intersection, all not satisfied
+                    op = OP_NULL;
+                    return OLAP_SUCCESS;
+                }
+            } else {
+                DCHECK_EQ(other.op, OP_IN) << "op: " << op << ", other.op: " << other.op;
+                if (other.operand_set.find(operand_field) == other.operand_set.end()) {
+                    // No intersection, all not satisfied
+                    op = OP_NULL;
+                    return OLAP_SUCCESS;
+                }
+            }
+            return OLAP_SUCCESS;
+        case OP_NE:
+            if (other.op == OP_NE) {
+                int cmp = operand_field->field()->compare_cell(*operand_field, *(other.operand_field));
+                if (cmp != 0) {
+                    // Transfer to OP_NOT_IN if they OP_NE to two different values
+                    op = OP_NOT_IN;
+                    operand_set.insert(operand_field);
+                    operand_set.insert(other.operand_field);
+                    min_value_field = cmp < 0 ? operand_field : other.operand_field;
+                    max_value_field = cmp > 0 ? operand_field : other.operand_field;
+                    // Invalidate operand_field after transferring to operand_set
+                    operand_field = nullptr;
+                }
+            } else {
+                DCHECK_EQ(other.op, OP_NOT_IN) << "op: " << op << ", other.op: " << other.op;
+                if (other.operand_set.size() == 1 &&
+                    operand_field->field()->compare_cell(*operand_field, *(other.min_value_field)) == 0) {
+                    // Do nothing if the other's only one value equal to operand_field
+                    return OLAP_SUCCESS;
+                }
+                // Transfer to OP_NOT_IN otherwise
+                op = OP_NOT_IN;
+                operand_set = other.operand_set;
+                min_value_field = other.min_value_field;
+                max_value_field = other.max_value_field;
+
+                if (operand_set.find(operand_field) != operand_set.end()) {
+                    // Exist a same value in operand_set, do nothing but release and invalidate operand_field
+                    operand_field = nullptr;
+                    return OLAP_SUCCESS;
+                }
+
+                // Insert and update min & max
+                operand_set.insert(operand_field);
+                if (operand_field->field()->compare_cell(*operand_field, *(min_value_field)) < 0) {
+                    min_value_field = operand_field;
+                }
+                if (operand_field->field()->compare_cell(*operand_field, *(max_value_field)) > 0) {
+                    max_value_field = operand_field;
+                }
+
+                // Invalidate operand_field after inserting to operand_set
+                operand_field = nullptr;
+            }
+            return OLAP_SUCCESS;
+        case OP_LT:
+        case OP_LE: {
+            int cmp = operand_field->field()->compare_cell(*operand_field, *(other.operand_field));
+            if (op == other.op) {
+                if (cmp > 0) {
+                    operand_field = other.operand_field;
+                }
+                return OLAP_SUCCESS;
+            }
+            if (cmp == 0) {
+                op = OP_LT;
+            }
+            return OLAP_SUCCESS;
+        }
+        case OP_GT:
+        case OP_GE: {
+            int cmp = operand_field->field()->compare_cell(*operand_field, *(other.operand_field));
+            if (op == other.op) {
+                if (cmp < 0) {
+                    operand_field = other.operand_field;
+                }
+                return OLAP_SUCCESS;
+            }
+            if (cmp == 0) {
+                op = OP_GT;
+            }
+            return OLAP_SUCCESS;
+        }
+        case OP_IN:
+            if (other.op == OP_IN) {
+                for (auto operand = operand_set.begin(); operand != operand_set.end();) {
+                    if (other.operand_set.find(*operand) == other.operand_set.end()) {
+                        // Not in other's operand_set, invalidate and release it
+                        operand = operand_set.erase(operand);
+                    } else {
+                        ++operand;
+                    }
+                }
+                if (operand_set.empty()) {
+                    // No intersection, all not satisfied
+                    op = OP_NULL;
+                    return OLAP_SUCCESS;
+                }
+
+                min_value_field = nullptr;
+                max_value_field = nullptr;
+                if (operand_set.size() == 1) {
+                    // Transfer to OP_EQ
+                    op = OP_EQ;
+                    operand_field = *operand_set.begin();
+                    operand_set.clear();
+                    return OLAP_SUCCESS;
+                }
+
+                // Update min & max
+                for (const auto& operand : operand_set) {
+                    if (min_value_field == nullptr || operand->field()->compare_cell(*min_value_field, *(operand)) > 0) {
+                        min_value_field = operand;
+                    }
+                    if (max_value_field == nullptr || operand->field()->compare_cell(*max_value_field, *(operand)) < 0) {
+                        max_value_field = operand;
+                    }
+                }
+            } else {
+                DCHECK_EQ(other.op, OP_EQ) << "op: " << op << ", other.op: " << other.op;
+                if (operand_set.find(other.operand_field) == operand_set.end()) {
+                    // No intersection, all not satisfied
+                    op = OP_NULL;
+                    return OLAP_SUCCESS;
+                }
+
+                // Transfer to OP_EQ
+                op = OP_EQ;
+                operand_field = other.operand_field;
+
+                // Invalidate
+                operand_set.clear();
+                min_value_field = nullptr;
+                max_value_field = nullptr;
+            }
+            return OLAP_SUCCESS;
+        case OP_NOT_IN:
+            if (other.op == OP_NOT_IN) {
+                // Update min & max
+                if (min_value_field->field()->compare_cell(*min_value_field, *(other.min_value_field)) > 0) {
+                    min_value_field = other.min_value_field;
+                }
+                if (max_value_field->field()->compare_cell(*max_value_field, *(other.max_value_field)) < 0) {
+                    max_value_field = other.max_value_field;
+                }
+                // Update operand_set
+                operand_set.insert(other.operand_set.begin(), other.operand_set.end());
+            } else {
+                DCHECK_EQ(other.op, OP_NE) << "op: " << op << ", other.op: " << other.op;
+                if (operand_set.find(other.operand_field) != operand_set.end()) {
+                    // Exist a same value in operand_set, do nothing but release and invalidate this operand
+                    return OLAP_SUCCESS;
+                }
+
+                // Update min & max
+                if (other.operand_field->field()->compare_cell(*min_value_field, *(other.operand_field)) > 0) {
+                    min_value_field = other.operand_field;
+                }
+                if (other.operand_field->field()->compare_cell(*max_value_field, *(other.operand_field)) < 0) {
+                    max_value_field = other.operand_field;
+                }
+
+                // Update operand_set
+                operand_set.insert(other.operand_field);
+            }
+            return OLAP_SUCCESS;
+        case OP_IS:
+            if (operand_field->is_null() != other.operand_field->is_null()) {
+                // No intersection, all not satisfied
+                op = OP_NULL;
+                return OLAP_SUCCESS;
+            }
+            return OLAP_SUCCESS;
+        default:
+            op = OP_ALL;;
+            return OLAP_ERR_READER_INITIALIZE_ERROR;
+    }
+}
+
+OLAPStatus Cond::union_cond(const Cond& other) {
+    DCHECK(op == other.op ||
+           (op == OP_NOT_IN && other.op == OP_NE) ||
+           (op == OP_NE && other.op == OP_NOT_IN) ||
+           (op == OP_IN && other.op == OP_EQ) ||
+           (op == OP_EQ && other.op == OP_IN) ||
+           (op == OP_LT && other.op == OP_LE) ||
+           (op == OP_LE && other.op == OP_LT) ||
+           (op == OP_GT && other.op == OP_GE) ||
+           (op == OP_GE && other.op == OP_GT)) << "op: " << op << ", other.op: " << other.op;
+    switch (op) {
+        case OP_EQ:
+            if (other.op == OP_EQ) {
+                int cmp = operand_field->field()->compare_cell(*operand_field, *(other.operand_field));
+                if (cmp != 0) {
+                    // Transfer to OP_IN if they OP_EQ to two different values
+                    op = OP_IN;
+                    operand_set.insert(operand_field);
+                    operand_set.insert(other.operand_field);
+                    min_value_field = cmp < 0 ? operand_field : other.operand_field;
+                    max_value_field = cmp > 0 ? operand_field : other.operand_field;
+                    // Invalidate operand_field after transferring to operand_set
+                    operand_field = nullptr;
+                }
+            } else {
+                DCHECK_EQ(other.op, OP_IN) << "op: " << op << ", other.op: " << other.op;
+                // Transfer to OP_IN
+                op = OP_IN;
+                operand_set = other.operand_set;
+                min_value_field = other.min_value_field;
+                max_value_field = other.max_value_field;
+
+                if (operand_set.find(operand_field) == operand_set.end()) {
+                    // Insert and update min & max
+                    operand_set.insert(operand_field);
+                    if (operand_field->field()->compare_cell(*operand_field, *(min_value_field)) < 0) {
+                        min_value_field = operand_field;
+                    }
+                    if (operand_field->field()->compare_cell(*operand_field, *(max_value_field)) > 0) {
+                        max_value_field = operand_field;
+                    }
+                }
+                operand_field = nullptr;
+            }
+            return OLAP_SUCCESS;
+        case OP_NE:
+            if (other.op == OP_NE) {
+                int cmp = operand_field->field()->compare_cell(*operand_field, *(other.operand_field));
+                if (cmp != 0) {
+                    // All satisfied
+                    op = OP_ALL;
+                    operand_field = nullptr;
+                }
+            } else {
+                DCHECK_EQ(other.op, OP_NOT_IN) << "op: " << op << ", other.op: " << other.op;
+                if (other.operand_set.find(operand_field) == other.operand_set.end()) {
+                    // All satisfied
+                    op = OP_ALL;
+                    operand_field = nullptr;
+                }
+            }
+            return OLAP_SUCCESS;
+        case OP_LT:
+        case OP_LE: {
+            int cmp = operand_field->field()->compare_cell(*operand_field, *(other.operand_field));
+            if (op == other.op) {
+                if (cmp < 0) {
+                    operand_field = other.operand_field;
+                }
+                return OLAP_SUCCESS;
+            }
+            if (cmp == 0) {
+                op = OP_LE;
+            }
+            return OLAP_SUCCESS;
+        }
+        case OP_GT:
+        case OP_GE: {
+            int cmp = operand_field->field()->compare_cell(*operand_field, *(other.operand_field));
+            if (op == other.op) {
+                if (cmp > 0) {
+                    operand_field = other.operand_field;
+                }
+                return OLAP_SUCCESS;
+            }
+            if (cmp == 0) {
+                op = OP_GE;
+            }
+            return OLAP_SUCCESS;
+        }
+        case OP_IN:
+            if (other.op == OP_IN) {
+                for (const auto& operand : other.operand_set) {
+                    if (operand_set.find(operand) == operand_set.end()) {
+                        operand_set.insert(operand);
+                        if (operand->field()->compare_cell(*min_value_field, *operand) > 0) {
+                            min_value_field = operand;
+                        }
+                        if (operand->field()->compare_cell(*max_value_field, *operand) < 0) {
+                            max_value_field = operand;
+                        }
+                    }
+                }
+            } else {
+                DCHECK_EQ(other.op, OP_EQ) << "op: " << op << ", other.op: " << other.op;
+                if (operand_set.find(other.operand_field) == operand_set.end()) {
+                    operand_set.insert(other.operand_field);
+                    if (other.operand_field->field()->compare_cell(*min_value_field, *(other.operand_field)) > 0) {
+                        min_value_field = other.operand_field;
+                    }
+                    if (other.operand_field->field()->compare_cell(*max_value_field, *(other.operand_field)) < 0) {
+                        max_value_field = other.operand_field;
+                    }
+                }
+            }
+            return OLAP_SUCCESS;
+        case OP_NOT_IN:
+            if (other.op == OP_NOT_IN) {
+                for (auto operand = operand_set.begin(); operand != operand_set.end();) {
+                    if (other.operand_set.find(*operand) != other.operand_set.end()) {
+                        ++operand;
+                    } else {
+                        operand = operand_set.erase(operand);
+                    }
+                }
+                min_value_field = nullptr;
+                max_value_field = nullptr;
+                if (operand_set.empty()) {
+                    // All satisfied
+                    op = OP_ALL;
+                    return OLAP_SUCCESS;
+                }
+
+                if (operand_set.size() == 1) {
+                    // Transfer to OP_NE
+                    op = OP_NE;
+                    operand_field = *operand_set.begin();
+                    operand_set.clear();
+                    return OLAP_SUCCESS;
+                }
+
+                // Update min & max
+                for (const auto& operand : operand_set) {
+                    if (min_value_field == nullptr || operand->field()->compare_cell(*min_value_field, *(operand)) > 0) {
+                        min_value_field = operand;
+                    }
+                    if (max_value_field == nullptr || operand->field()->compare_cell(*max_value_field, *(operand)) < 0) {
+                        max_value_field = operand;
+                    }
+                }
+            } else {
+                DCHECK_EQ(other.op, OP_NE) << "op: " << op << ", other.op: " << other.op;
+                min_value_field = nullptr;
+                max_value_field = nullptr;
+                if (operand_set.find(other.operand_field) == operand_set.end()) {
+                    // All satisfied
+                    op = OP_ALL;
+                    operand_set.clear();
+                    return OLAP_SUCCESS;
+                }
+
+                // Transfer to OP_NE
+                op = OP_NE;
+                operand_field = other.operand_field;
+                operand_set.clear();
+            }
+            return OLAP_SUCCESS;
+        case OP_IS:
+            if (operand_field->is_null() != other.operand_field->is_null()) {
+                // All satisfied
+                op = OP_ALL;
+                operand_field = nullptr;
+                return OLAP_SUCCESS;
+            }
+            return OLAP_SUCCESS;
+        default:
+            op = OP_ALL;;
+            return OLAP_ERR_READER_INITIALIZE_ERROR;
+    }
+}
+
 bool Cond::eval(const RowCursorCell& cell) const {
     if (cell.is_null() && op != OP_IS) {
         //任何非OP_IS operand和NULL的运算都是false
@@ -194,15 +559,15 @@ bool Cond::eval(const RowCursorCell& cell) const {
     case OP_GE:
         return operand_field->field()->compare_cell(*operand_field, cell) <= 0;
     case OP_IN: {
-        WrapperField wrapperField(const_cast<Field*>(min_value_field->field()), cell);
-        auto ret = operand_set.find(&wrapperField) != operand_set.end();
-        wrapperField.release_field();
+        auto wrapperField = std::make_shared<WrapperField>(const_cast<Field*>(min_value_field->field()), cell);
+        auto ret = operand_set.find(wrapperField) != operand_set.end();
+        wrapperField->release_field();
         return ret;
     }
     case OP_NOT_IN: {
-        WrapperField wrapperField(const_cast<Field*>(min_value_field->field()), cell);
-        auto ret = operand_set.find(&wrapperField) == operand_set.end();
-        wrapperField.release_field();
+        auto wrapperField = std::make_shared<WrapperField>(const_cast<Field*>(min_value_field->field()), cell);
+        auto ret = operand_set.find(wrapperField) == operand_set.end();
+        wrapperField->release_field();
         return ret;
     }
     case OP_IS: {
@@ -350,7 +715,7 @@ int Cond::del_eval(const std::pair<WrapperField*, WrapperField*>& stat) const {
     }
     case OP_IN: {
         if (stat.first->cmp(stat.second) == 0) {
-            if (operand_set.find(stat.first) != operand_set.end()) {
+            if (operand_set.find(std::shared_ptr<WrapperField>(stat.first, [](WrapperField*){})) != operand_set.end()) {
                 ret = DEL_SATISFIED;
             } else {
                 ret = DEL_NOT_SATISFIED;
@@ -366,7 +731,7 @@ int Cond::del_eval(const std::pair<WrapperField*, WrapperField*>& stat) const {
     }
     case OP_NOT_IN: {
         if (stat.first->cmp(stat.second) == 0) {
-            if (operand_set.find(stat.first) == operand_set.end()) {
+            if (operand_set.find(std::shared_ptr<WrapperField>(stat.first, [](WrapperField*){})) == operand_set.end()) {
                 ret = DEL_SATISFIED;
             } else {
                 ret = DEL_NOT_SATISFIED;
@@ -492,23 +857,74 @@ bool Cond::eval(const segment_v2::BloomFilter* bf) const {
     return true;
 }
 
-CondColumn::~CondColumn() {
-    for (auto& it : _conds) {
-        delete it;
+int Cond::eval_cost() const {
+    switch (op) {
+        case OP_NULL:
+        case OP_ALL:
+            return 0;
+        case OP_IS:
+            return 1;
+        case OP_EQ:
+        case OP_NE:
+        case OP_LT:
+        case OP_LE:
+        case OP_GT:
+        case OP_GE:
+            return 2 * operand_field->field_size();
+        case OP_IN:
+        case OP_NOT_IN:
+            return 3 * operand_set.size() * (*operand_set.begin())->field_size();
+        default:
+            return 0;
     }
 }
 
 // PRECONDITION 1. index is valid; 2. at least has one operand
 OLAPStatus CondColumn::add_cond(const TCondition& tcond, const TabletColumn& column) {
-    std::unique_ptr<Cond> cond(new Cond());
+    auto cond = std::make_shared<Cond>();
     auto res = cond->init(tcond, column);
     if (res != OLAP_SUCCESS) {
         return res;
     }
-    _conds.push_back(cond.release());
+    _conds.push_back(cond);
     return OLAP_SUCCESS;
 }
 
+void CondColumn::merge_cond(const CondColumn& cond_col) {
+    DCHECK_EQ(_is_key, cond_col._is_key);
+    DCHECK_EQ(_col_index, cond_col._col_index);
+
+    for (auto& cond : _conds) {
+        for (const auto& old_cond : cond_col._conds) {
+            if ((cond->op == old_cond->op) ||
+                (cond->op == OP_NOT_IN && old_cond->op == OP_NE) ||
+                (cond->op == OP_NE && old_cond->op == OP_NOT_IN) ||
+                (cond->op == OP_IN && old_cond->op == OP_EQ) ||
+                (cond->op == OP_EQ && old_cond->op == OP_IN) ||
+                (cond->op == OP_LT && old_cond->op == OP_LE) ||
+                (cond->op == OP_LE && old_cond->op == OP_LT) ||
+                (cond->op == OP_GT && old_cond->op == OP_GE) ||
+                (cond->op == OP_GE && old_cond->op == OP_GT)) {
+                CHECK_EQ(cond->union_cond(*old_cond), OLAP_SUCCESS);
+                break;
+            }
+        }
+    }
+}
+
+void CondColumn::normalize() {
+    // Do lower cost evaluation at first.
+    std::sort(_conds.begin(), _conds.end(),
+         [] (const std::shared_ptr<Cond>& left,
+             const std::shared_ptr<Cond>& right) {
+             return left->eval_cost() < right->eval_cost();
+         });
+    _cost = 0.0;
+    for (const auto& cond : _conds) {
+        _cost += cond->eval_cost();
+    }
+}
+
 bool CondColumn::eval(const RowCursor& row) const {
     auto cell = row.cell(_col_index);
     for (auto& each_cond : _conds) {
@@ -595,16 +1011,17 @@ OLAPStatus Conditions::append_condition(const TCondition& tcond) {
     }
 
     // Skip column which is non-key, or whose type is string or float
+    // TODO(yingchun): why skip?
     const TabletColumn& column = _schema->column(index);
     if (column.type() == OLAP_FIELD_TYPE_DOUBLE || column.type() == OLAP_FIELD_TYPE_FLOAT) {
         return OLAP_SUCCESS;
     }
 
     CondColumn* cond_col = nullptr;
-    auto it = _columns.find(index);
-    if (it == _columns.end()) {
+    auto it = _index_conds.find(index);
+    if (it == _index_conds.end()) {
         cond_col = new CondColumn(*_schema, index);
-        _columns[index] = cond_col;
+        _index_conds[index] = cond_col;
     } else {
         cond_col = it->second;
     }
@@ -612,19 +1029,51 @@ OLAPStatus Conditions::append_condition(const TCondition& tcond) {
     return cond_col->add_cond(tcond, column);
 }
 
+void Conditions::normalize() {
+    for (const auto& cond_column : _index_conds) {
+        _sorted_conds.push_back(cond_column.second);
+    }
+
+    // Do lower cost evaluation at first.
+    std::sort(_sorted_conds.begin(), _sorted_conds.end(),
+              [] (CondColumn* left,
+                  CondColumn* right) {
+                  return left->eval_cost() < right->eval_cost();
+              });
+    _cost = 0.0;
+    for (const auto& cond : _sorted_conds) {
+        _cost += cond->eval_cost();
+    }
+}
+
+void Conditions::merge_condition(const std::vector<CondColumn*>& cond_cols) {
+    for (const auto& cond_col : cond_cols) {
+        int32_t index = cond_col->col_index();
+        auto it = _index_conds.find(index);
+        if (it == _index_conds.end()) {
+            CondColumn* old_cond_col = new CondColumn(*_schema, index);
+            old_cond_col->_conds = cond_col->conds();
+            _index_conds[index] = old_cond_col;
+        } else {
+            it->second->merge_cond(*cond_col);
+        }
+    }
+}
+
 bool Conditions::delete_conditions_eval(const RowCursor& row) const {
-    if (_columns.empty()) {
+    if (_sorted_conds.empty()) {
         return false;
     }
-
-    for (auto& each_cond : _columns) {
-        if (_cond_column_is_key_or_duplicate(each_cond.second) && !each_cond.second->eval(row)) {
+    
+    for (auto& each_cond : _sorted_conds) {
+        // TODO(yingchun): why only judge key and dup?
+        if (_cond_column_is_key_or_duplicate(each_cond) && !each_cond->eval(row)) {
             return false;
         }
     }
 
     VLOG(3) << "Row meets the delete conditions. "
-            << "condition_count=" << _columns.size() << ", row=" << row.to_string();
+            << "condition_count=" << _sorted_conds.size() << ", row=" << row.to_string();
     return true;
 }
 
@@ -632,10 +1081,10 @@ bool Conditions::rowset_pruning_filter(const std::vector<KeyRange>& zone_maps) c
     // ZoneMap will store min/max of rowset.
     // The function is to filter rowset using ZoneMaps
     // and query predicates.
-    for (auto& cond_it : _columns) {
-        if (_cond_column_is_key_or_duplicate(cond_it.second)) {
-            if (cond_it.first < zone_maps.size() &&
-                !cond_it.second->eval(zone_maps.at(cond_it.first))) {
+    for (auto& cond_it : _sorted_conds) {
+        if (_cond_column_is_key_or_duplicate(cond_it)) {
+            if (cond_it->col_index() < zone_maps.size() &&
+                !cond_it->eval(zone_maps.at(cond_it->col_index()))) {
                 return true;
             }
         }
@@ -644,9 +1093,10 @@ bool Conditions::rowset_pruning_filter(const std::vector<KeyRange>& zone_maps) c
 }
 
 int Conditions::delete_pruning_filter(const std::vector<KeyRange>& zone_maps) const {
-    if (_columns.empty()) {
+    if (_sorted_conds.empty()) {
         return DEL_NOT_SATISFIED;
     }
+
     // ZoneMap and DeletePredicate are all stored in TabletMeta.
     // This function is to filter rowset using ZoneMap and Delete Predicate.
     /*
@@ -658,19 +1108,19 @@ int Conditions::delete_pruning_filter(const std::vector<KeyRange>& zone_maps) co
     int ret = DEL_NOT_SATISFIED;
     bool del_partial_satisfied = false;
     bool del_not_satisfied = false;
-    for (auto& cond_it : _columns) {
+    for (auto& cond_it : _sorted_conds) {
         /*
          * this is base on the assumption that the delete condition
          * is only about key field, not about value field except the storage model is duplicate.
         */
-        if (_cond_column_is_key_or_duplicate(cond_it.second) && cond_it.first > zone_maps.size()) {
+        if (!_cond_column_is_key_or_duplicate(cond_it) || cond_it->col_index() > zone_maps.size()) {
             LOG(WARNING) << "where condition not equal column statistics size. "
-                         << "cond_id=" << cond_it.first << ", zone_map_size=" << zone_maps.size();
+                         << "cond_id=" << cond_it->col_index() << ", zone_map_size=" << zone_maps.size();
             del_partial_satisfied = true;
             continue;
         }
 
-        int del_ret = cond_it.second->del_eval(zone_maps.at(cond_it.first));
+        int del_ret = cond_it->del_eval(zone_maps.at(cond_it->col_index()));
         if (DEL_SATISFIED == del_ret) {
             continue;
         } else if (DEL_PARTIAL_SATISFIED == del_ret) {
@@ -692,8 +1142,8 @@ int Conditions::delete_pruning_filter(const std::vector<KeyRange>& zone_maps) co
 }
 
 CondColumn* Conditions::get_column(int32_t cid) const {
-    auto iter = _columns.find(cid);
-    if (iter != _columns.end()) {
+    auto iter = _index_conds.find(cid);
+    if (iter != _index_conds.end()) {
         return iter->second;
     }
     return nullptr;
diff --git a/be/src/olap/olap_cond.h b/be/src/olap/olap_cond.h
index 54d699e51f60b8..397df7e9656cae 100644
--- a/be/src/olap/olap_cond.h
+++ b/be/src/olap/olap_cond.h
@@ -38,27 +38,30 @@ class WrapperField;
 class RowCursorCell;
 
 enum CondOp {
-    OP_NULL = -1, // invalid op
-    OP_EQ = 0,    // equal
-    OP_NE = 1,    // not equal
-    OP_LT = 2,    // less than
-    OP_LE = 3,    // less or equal
-    OP_GT = 4,    // greater than
-    OP_GE = 5,    // greater or equal
-    OP_IN = 6,    // in
-    OP_IS = 7,    // is null or not null
-    OP_NOT_IN = 8 // not in
+    OP_NULL = -1,   // invalid op
+    OP_EQ = 0,      // equal
+    OP_NE = 1,      // not equal
+    OP_LT = 2,      // less than
+    OP_LE = 3,      // less or equal
+    OP_GT = 4,      // greater than
+    OP_GE = 5,      // greater or equal
+    OP_IN = 6,      // in
+    OP_IS = 7,      // is null or not null
+    OP_NOT_IN = 8,  // not in
+    OP_ALL = 100    // all satisfied
 };
 
 // Hash functor for IN set
 struct FieldHash {
-    size_t operator()(const WrapperField* field) const { return field->hash_code(); }
+    size_t operator()(const std::shared_ptr<WrapperField>& field) const {
+        return field->hash_code();
+    }
 };
 
 // Equal function for IN set
 struct FieldEqual {
-    bool operator()(const WrapperField* left, const WrapperField* right) const {
-        return left->cmp(right) == 0;
+    bool operator()(const std::shared_ptr<WrapperField>& left, const std::shared_ptr<WrapperField>& right) const {
+        return left->cmp(right.get()) == 0;
     }
 };
 
@@ -66,10 +69,14 @@ struct FieldEqual {
 struct Cond {
 public:
     Cond() = default;
-    ~Cond();
 
     OLAPStatus init(const TCondition& tcond, const TabletColumn& column);
 
+    // 'other' may be invalid after intersection
+    OLAPStatus intersection_cond(const Cond& other);
+
+    OLAPStatus union_cond(const Cond& other);
+
     // 用一行数据的指定列同条件进行比较，如果符合过滤条件，
     // 即按照此条件，行应被过滤掉，则返回true，否则返回false
     bool eval(const RowCursorCell& cell) const;
@@ -84,15 +91,17 @@ struct Cond {
 
     bool can_do_bloom_filter() const { return op == OP_EQ || op == OP_IN || op == OP_IS; }
 
+    int eval_cost() const;
+
     CondOp op = OP_NULL;
     // valid when op is not OP_IN and OP_NOT_IN
-    WrapperField* operand_field = nullptr;
+    std::shared_ptr<WrapperField> operand_field;
     // valid when op is OP_IN or OP_NOT_IN
-    typedef std::unordered_set<const WrapperField*, FieldHash, FieldEqual> FieldSet;
+    typedef std::unordered_set<std::shared_ptr<WrapperField>, FieldHash, FieldEqual> FieldSet;    // TODO(yingchun): should use shared_ptr
     FieldSet operand_set;
     // valid when op is OP_IN or OP_NOT_IN, represents the minimum or maximum value of in elements
-    WrapperField* min_value_field = nullptr;
-    WrapperField* max_value_field = nullptr;
+    std::shared_ptr<WrapperField> min_value_field;
+    std::shared_ptr<WrapperField> max_value_field;
 };
 
 // 所有归属于同一列上的条件二元组，聚合在一个CondColumn上
@@ -101,9 +110,12 @@ class CondColumn {
     CondColumn(const TabletSchema& tablet_schema, int32_t index) : _col_index(index) {
         _is_key = tablet_schema.column(_col_index).is_key();
     }
-    ~CondColumn();
+    ~CondColumn() = default;
 
     OLAPStatus add_cond(const TCondition& tcond, const TabletColumn& column);
+    void merge_cond(const CondColumn& cond_col);
+
+    void normalize();
 
     // 对一行数据中的指定列，用所有过滤条件进行比较，如果所有条件都满足，则过滤此行
     // Return true means this row should be filtered out, otherwise return false
@@ -134,15 +146,27 @@ class CondColumn {
 
     inline bool is_key() const { return _is_key; }
 
-    const std::vector<Cond*>& conds() const { return _conds; }
+    const std::vector<std::shared_ptr<Cond>>& conds() const {
+        return _conds;
+    }
+
+    double eval_cost() const {
+        return _cost;
+    }
+
+    int32_t col_index() const {
+        return _col_index;
+    }
 
 private:
     friend class Conditions;
 
     bool _is_key = false;
     int32_t _col_index = 0;
+    // TODO(yingchun): DELETE FROM xx WHERE col1 > a AND col1 < b
     // Conds in _conds are in 'AND' relationship
-    std::vector<Cond*> _conds;
+    std::vector<std::shared_ptr<Cond>> _conds;
+    double _cost = 0.0;
 };
 
 // 一次请求所关联的条件
@@ -156,10 +180,10 @@ class Conditions {
     ~Conditions() { finalize(); }
 
     void finalize() {
-        for (auto& it : _columns) {
+        for (auto& it : _index_conds) {
             delete it.second;
         }
-        _columns.clear();
+        _index_conds.clear();
     }
 
     // TODO(yingchun): should do it in constructor
@@ -170,7 +194,11 @@ class Conditions {
     // 1. column不属于key列
     // 2. column类型是double, float
     OLAPStatus append_condition(const TCondition& condition);
-    
+
+    void normalize();
+
+    void merge_condition(const std::vector<CondColumn*>& cond_cols);
+
     // 通过所有列上的删除条件对RowCursor进行过滤
     // Return true means this row should be filtered out, otherwise return false
     bool delete_conditions_eval(const RowCursor& row) const;
@@ -181,10 +209,28 @@ class Conditions {
     // Whether the rowset satisfied delete condition
     int delete_pruning_filter(const std::vector<KeyRange>& zone_maps) const;
 
-    const CondColumns& columns() const { return _columns; }
+    const CondColumns& index_conds() const {
+        return _index_conds;
+    }
+
+    CondColumn* col_cond(int32_t col_index) const {
+        auto col_cond = _index_conds.find(col_index);
+        if (col_cond == _index_conds.end()) {
+            return nullptr;
+        }
+        return col_cond->second;
+    }
+
+    const std::vector<CondColumn*>& sorted_conds() const {
+        return _sorted_conds;
+    }
 
     CondColumn* get_column(int32_t cid) const;
 
+    double eval_cost() const {
+        return _cost;
+    }
+
 private:
     bool _cond_column_is_key_or_duplicate(const CondColumn* cc) const {
         return cc->is_key() || _schema->keys_type() == KeysType::DUP_KEYS;
@@ -192,8 +238,13 @@ class Conditions {
 
 private:
     const TabletSchema* _schema = nullptr;
+
+    // TODO(yingchun): DELETE FROM xx WHERE col1 IN (a) AND col2 IN (b)
     // CondColumns in _index_conds are in 'AND' relationship
-    CondColumns _columns; // list of condition column
+    CondColumns _index_conds;
+
+    double _cost = 0.0;
+    std::vector<CondColumn*> _sorted_conds;
 
     DISALLOW_COPY_AND_ASSIGN(Conditions);
 };
diff --git a/be/src/olap/reader.cpp b/be/src/olap/reader.cpp
index 0b05c744208de7..3bd3ddb1b832e0 100644
--- a/be/src/olap/reader.cpp
+++ b/be/src/olap/reader.cpp
@@ -401,6 +401,7 @@ OLAPStatus Reader::_capture_rs_readers(const ReaderParams& read_params) {
 
 OLAPStatus Reader::_init_params(const ReaderParams& read_params) {
     read_params.check_validation();
+    LOG(WARNING) << read_params.to_string();
 
     _aggregation = read_params.aggregation;
     _need_agg_finalize = read_params.need_agg_finalize;
@@ -453,10 +454,11 @@ OLAPStatus Reader::_init_return_columns(const ReaderParams& read_params) {
         if (!_delete_handler.empty() && read_params.aggregation) {
             set<uint32_t> column_set(_return_columns.begin(), _return_columns.end());
             for (const auto& conds : _delete_handler.get_delete_conditions()) {
-                for (const auto& cond_column : conds.del_cond->columns()) {
-                    if (column_set.find(cond_column.first) == column_set.end()) {
-                        column_set.insert(cond_column.first);
-                        _return_columns.push_back(cond_column.first);
+                for (const auto& cond_column : conds.del_cond->sorted_conds()) {
+                    auto cid = cond_column->col_index();
+                    if (column_set.find(cid) == column_set.end()) {
+                        column_set.insert(cid);
+                        _return_columns.push_back(cid);
                     }
                 }
             }
@@ -500,8 +502,8 @@ OLAPStatus Reader::_init_return_columns(const ReaderParams& read_params) {
 
 void Reader::_init_seek_columns() {
     std::unordered_set<uint32_t> column_set(_return_columns.begin(), _return_columns.end());
-    for (auto& it : _conditions.columns()) {
-        column_set.insert(it.first);
+    for (auto& it : _conditions.sorted_conds()) {
+        column_set.insert(it->col_index());
     }
     size_t max_key_column_count = 0;
     for (const auto& key : _keys_param.start_keys) {
@@ -590,6 +592,7 @@ void Reader::_init_conditions_param(const ReaderParams& read_params) {
             }
         }
     }
+    _conditions.normalize();
 }
 
 #define COMPARISON_PREDICATE_CONDITION_VALUE(NAME, PREDICATE)                              \
@@ -835,14 +838,14 @@ ColumnPredicate* Reader::_parse_to_predicate(const TCondition& condition) {
 
 void Reader::_init_load_bf_columns(const ReaderParams& read_params) {
     // add all columns with condition to _load_bf_columns
-    for (const auto& cond_column : _conditions.columns()) {
-        if (!_tablet->tablet_schema().column(cond_column.first).is_bf_column()) {
+    for (const auto& cond_column : _conditions.sorted_conds()) {
+        if (!_tablet->tablet_schema().column(cond_column->col_index()).is_bf_column()) {
             continue;
         }
-        for (const auto& cond : cond_column.second->conds()) {
+        for (const auto& cond : cond_column->conds()) {
             if (cond->op == OP_EQ ||
                 (cond->op == OP_IN && cond->operand_set.size() < MAX_OP_IN_FIELD_NUM)) {
-                _load_bf_columns.insert(cond_column.first);
+                _load_bf_columns.insert(cond_column->col_index());
             }
         }
     }
diff --git a/be/src/olap/reader.h b/be/src/olap/reader.h
index 9da3006252a4eb..28f28ccce67307 100644
--- a/be/src/olap/reader.h
+++ b/be/src/olap/reader.h
@@ -182,7 +182,7 @@ class Reader {
     bool _need_agg_finalize = true;
     ReaderType _reader_type = READER_QUERY;
     bool _next_delete_flag = false;
-    bool _filter_delete = false;
+    bool _filter_delete = false;  // Whether to delete the filtered rows
     bool _has_sequence_col = false;
     int32_t _sequence_col_idx = -1;
     const RowCursor* _next_key = nullptr;
diff --git a/be/src/olap/rowset/segment_reader.cpp b/be/src/olap/rowset/segment_reader.cpp
index 80f033611984ae..0d6a74fe78b0c2 100644
--- a/be/src/olap/rowset/segment_reader.cpp
+++ b/be/src/olap/rowset/segment_reader.cpp
@@ -394,14 +394,14 @@ OLAPStatus SegmentReader::_pick_delete_row_groups(uint32_t first_block, uint32_t
 
             bool del_partial_satisfied = false;
             bool del_not_satisfied = false;
-            for (auto& i : delete_condition.del_cond->columns()) {
-                ColumnId table_column_id = i.first;
+            for (auto& col_cond : delete_condition.del_cond->sorted_conds()) {
+                ColumnId table_column_id = col_cond->col_index();
                 ColumnId unique_column_id = _tablet_id_to_unique_id_map[table_column_id];
                 if (0 == _unique_id_to_segment_id_map.count(unique_column_id)) {
                     continue;
                 }
                 StreamIndexReader* index_reader = _indices[unique_column_id];
-                int del_ret = i.second->del_eval(index_reader->entry(j).column_statistic().pair());
+                int del_ret = col_cond->del_eval(index_reader->entry(j).column_statistic().pair());
                 if (DEL_SATISFIED == del_ret) {
                     continue;
                 } else if (DEL_PARTIAL_SATISFIED == del_ret) {
@@ -412,7 +412,7 @@ OLAPStatus SegmentReader::_pick_delete_row_groups(uint32_t first_block, uint32_t
                 }
             }
 
-            if (true == del_not_satisfied || 0 == delete_condition.del_cond->columns().size()) {
+            if (true == del_not_satisfied || delete_condition.del_cond->sorted_conds().empty()) {
                 //if state is DEL_PARTIAL_SATISFIED last_time, cannot be set as DEL_NOT_SATISFIED
                 //it is special for for delete condition
                 if (DEL_PARTIAL_SATISFIED == _include_blocks[j]) {
@@ -471,21 +471,21 @@ OLAPStatus SegmentReader::_pick_row_groups(uint32_t first_block, uint32_t last_b
 
     _pick_delete_row_groups(first_block, last_block);
 
-    if (NULL == _conditions || _conditions->columns().size() == 0) {
+    if (NULL == _conditions || _conditions->sorted_conds().empty()) {
         return OLAP_SUCCESS;
     }
 
     OlapStopWatch timer;
     timer.reset();
 
-    for (auto& i : _conditions->columns()) {
-        FieldAggregationMethod aggregation = _get_aggregation_by_index(i.first);
+    for (const auto& cond_col : _conditions->sorted_conds()) {
+        FieldAggregationMethod aggregation = _get_aggregation_by_index(cond_col->col_index());
         bool is_continue = (aggregation == OLAP_FIELD_AGGREGATION_NONE);
         if (!is_continue) {
             continue;
         }
 
-        ColumnId table_column_id = i.first;
+        ColumnId table_column_id = cond_col->col_index();
         ColumnId unique_column_id = _tablet_id_to_unique_id_map[table_column_id];
         if (0 == _unique_id_to_segment_id_map.count(unique_column_id)) {
             continue;
@@ -496,7 +496,7 @@ OLAPStatus SegmentReader::_pick_row_groups(uint32_t first_block, uint32_t last_b
                 continue;
             }
 
-            if (!i.second->eval(index_reader->entry(j).column_statistic().pair())) {
+            if (!cond_col->eval(index_reader->entry(j).column_statistic().pair())) {
                 _include_blocks[j] = DEL_SATISFIED;
                 --_remain_block;
 
@@ -535,7 +535,8 @@ OLAPStatus SegmentReader::_pick_row_groups(uint32_t first_block, uint32_t last_b
                 continue;
             }
 
-            if (!_conditions->columns().at(i)->eval(bf_reader->entry(j))) {
+            CondColumn* col_cond = _conditions->col_cond(i);
+            if (col_cond != nullptr && !col_cond->eval(bf_reader->entry(j))) {
                 _include_blocks[j] = DEL_SATISFIED;
                 --_remain_block;
                 if (j < _block_count - 1) {
diff --git a/be/src/olap/rowset/segment_v2/segment.cpp b/be/src/olap/rowset/segment_v2/segment.cpp
index 68ddb6df2635d7..44135cd7c7592d 100644
--- a/be/src/olap/rowset/segment_v2/segment.cpp
+++ b/be/src/olap/rowset/segment_v2/segment.cpp
@@ -59,13 +59,13 @@ Status Segment::new_iterator(const Schema& schema, const StorageReadOptions& rea
     read_options.stats->total_segment_number++;
     // trying to prune the current segment by segment-level zone map
     if (read_options.conditions != nullptr) {
-        for (auto& column_condition : read_options.conditions->columns()) {
-            int32_t column_id = column_condition.first;
+        for (auto& column_condition : read_options.conditions->sorted_conds()) {
+            int32_t column_id = column_condition->col_index();
             if (_column_readers[column_id] == nullptr ||
                 !_column_readers[column_id]->has_zone_map()) {
                 continue;
             }
-            if (!_column_readers[column_id]->match_condition(column_condition.second)) {
+            if (!_column_readers[column_id]->match_condition(column_condition)) {
                 // any condition not satisfied, return.
                 iter->reset(new EmptySegmentIterator(schema));
                 read_options.stats->filtered_segment_number++;
diff --git a/be/src/olap/rowset/segment_v2/segment_iterator.cpp b/be/src/olap/rowset/segment_v2/segment_iterator.cpp
index 1519f62e36b041..7c88ff6da54675 100644
--- a/be/src/olap/rowset/segment_v2/segment_iterator.cpp
+++ b/be/src/olap/rowset/segment_v2/segment_iterator.cpp
@@ -224,8 +224,8 @@ Status SegmentIterator::_get_row_ranges_by_column_conditions() {
 Status SegmentIterator::_get_row_ranges_from_conditions(RowRanges* condition_row_ranges) {
     std::set<int32_t> cids;
     if (_opts.conditions != nullptr) {
-        for (auto& column_condition : _opts.conditions->columns()) {
-            cids.insert(column_condition.first);
+        for (auto& column_condition : _opts.conditions->sorted_conds()) {
+            cids.insert(column_condition->col_index());
         }
     }
     // first filter data by bloom filter index
@@ -262,15 +262,15 @@ Status SegmentIterator::_get_row_ranges_from_conditions(RowRanges* condition_row
     // final filter data with delete conditions
     for (auto& delete_condition : _opts.delete_conditions) {
         RowRanges delete_condition_row_ranges = RowRanges::create_single(0);
-        for (auto& delete_column_condition : delete_condition->columns()) {
-            const int32_t cid = delete_column_condition.first;
+        for (auto& delete_column_condition : delete_condition->sorted_conds()) {
+            const int32_t cid = delete_column_condition->col_index();
             CondColumn* column_cond = nullptr;
             if (_opts.conditions != nullptr) {
                 column_cond = _opts.conditions->get_column(cid);
             }
             RowRanges single_delete_condition_row_ranges = RowRanges::create_single(num_rows());
             RETURN_IF_ERROR(_column_iterators[cid]->get_row_ranges_by_zone_map(
-                    column_cond, delete_column_condition.second,
+                    column_cond, delete_column_condition,
                     &single_delete_condition_row_ranges));
             RowRanges::ranges_union(delete_condition_row_ranges, single_delete_condition_row_ranges,
                                     &delete_condition_row_ranges);
diff --git a/be/test/exprs/hybrid_set_test.cpp b/be/test/exprs/hybrid_set_test.cpp
index d946afa7bce15d..e36e0e4a59dd37 100644
--- a/be/test/exprs/hybrid_set_test.cpp
+++ b/be/test/exprs/hybrid_set_test.cpp
@@ -326,7 +326,7 @@ TEST_F(HybridSetTest, string) {
     b.len = 5;
     ASSERT_FALSE(set->find(&b));
 }
-TEST_F(HybridSetTest, timestamp) {
+TEST_F(HybridSetTest, DISABLED_timestamp) {
     CpuInfo::init();
 
     HybridSetBase* set = HybridSetBase::create_set(TYPE_DATETIME);
diff --git a/be/test/olap/CMakeLists.txt b/be/test/olap/CMakeLists.txt
index ffc4507971ca19..db2538237394d3 100644
--- a/be/test/olap/CMakeLists.txt
+++ b/be/test/olap/CMakeLists.txt
@@ -50,6 +50,7 @@ ADD_BE_TEST(decimal12_test)
 ADD_BE_TEST(column_vector_test)
 ADD_BE_TEST(storage_types_test)
 ADD_BE_TEST(aggregate_func_test)
+ADD_BE_TEST(conditions_test)
 ADD_BE_TEST(rowset/segment_v2/bitshuffle_page_test)
 ADD_BE_TEST(rowset/segment_v2/plain_page_test)
 ADD_BE_TEST(rowset/segment_v2/binary_plain_page_test)
diff --git a/be/test/olap/conditions_test.cpp b/be/test/olap/conditions_test.cpp
new file mode 100644
index 00000000000000..a85c02655a2b5a
--- /dev/null
+++ b/be/test/olap/conditions_test.cpp
@@ -0,0 +1,241 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include "olap/olap_cond.h"
+#include "olap/tablet_schema_helper.h"
+
+namespace doris {
+
+class CondTest : public testing::Test {
+public:
+    CondTest() {
+        _tablet_column = create_int_key(_col);
+    }
+
+    void SetUp() override {
+        CpuInfo::init();
+    }
+
+    Cond create_cond(const std::string& op, const std::vector<std::string>& operands) {
+        TCondition tcond;
+        tcond.__set_column_name(std::to_string(_col));
+        tcond.__set_condition_op(op);
+        tcond.__set_condition_values(operands);
+
+        Cond cond;
+        cond.init(tcond, _tablet_column);
+
+        return cond;
+    }
+
+    void check_operand(Cond cond, CondOp op, const std::string& operand) {
+        if (op == OP_ALL) {
+            ASSERT_EQ(cond.operand_field, nullptr);
+            ASSERT_EQ(cond.operand_set.size(), 0);
+            ASSERT_EQ(cond.min_value_field, nullptr);
+            ASSERT_EQ(cond.max_value_field, nullptr);
+        } else if (op == OP_IS) {
+            ASSERT_NE(cond.operand_field, nullptr);
+            ASSERT_EQ(operand == "NULL", cond.operand_field->is_null());
+        } else {
+            ASSERT_NE(cond.operand_field, nullptr);
+            ASSERT_EQ(operand, cond.operand_field->to_string());
+        }
+    }
+
+    void check_operands(Cond cond, const std::set<std::string>& operands) {
+        ASSERT_EQ(cond.operand_field, nullptr);
+        ASSERT_NE(cond.min_value_field, nullptr);
+        ASSERT_NE(cond.max_value_field, nullptr);
+        ASSERT_EQ(cond.min_value_field->to_string(), *(operands.begin()));
+        ASSERT_EQ(cond.max_value_field->to_string(), *(operands.rbegin()));
+        ASSERT_EQ(cond.operand_set.size(), operands.size());
+        for (const auto& operand : cond.operand_set) {
+            ASSERT_EQ(operands.count(operand->to_string()), 1);
+        }
+    }
+
+    void calc_and_check(bool is_intersection,
+                        Cond cond1, Cond cond2, CondOp op,
+                        const std::string& operand = "",
+                        const std::set<std::string>& operands = {}) {
+        if (is_intersection) {
+            ASSERT_EQ(cond1.intersection_cond(cond2), OLAP_SUCCESS);
+        } else {
+            ASSERT_EQ(cond1.union_cond(cond2), OLAP_SUCCESS);
+        }
+        ASSERT_EQ(cond1.op, op);
+        if (op != OP_NULL) {
+            if (op == OP_IN || op == OP_NOT_IN) {
+                check_operands(cond1, operands);
+            } else {
+                check_operand(cond1, op, operand);
+            }
+        }
+    }
+
+private:
+    static const int32_t _col = 0;
+    TabletColumn _tablet_column;
+};
+
+TEST_F(CondTest, InitTest) {
+}
+
+TEST_F(CondTest, IntersectionCondsTest) {
+    struct {
+        std::string op1;
+        std::vector<std::string> operands1;
+        std::string op2;
+        std::vector<std::string> operands2;
+        CondOp result_op;
+        std::string result_operand;
+        std::set<std::string> result_operands;
+    } test_cases[] = {{"=",   {"1"},           "=",   {"1"},           OP_EQ,     "1",        {}},
+                      {"=",   {"1"},           "=",   {"2"},           OP_NULL,   "",         {}},
+                      {"=",   {"1"},           "*=",  {"1", "2", "3"}, OP_EQ,     "1",        {}},
+                      {"=",   {"1"},           "*=",  {"2", "3"},      OP_NULL,   "",         {}},
+
+                      {"!=",  {"1"},           "!=",  {"1"},           OP_NE,     "1",        {}},
+                      {"!=",  {"1"},           "!=",  {"2"},           OP_NOT_IN, "",         {"1", "2"}},
+                      {"!=",  {"1"},           "!*=", {"1"},           OP_NE,     "1",        {}},
+                      {"!=",  {"1"},           "!*=", {"1", "2", "3"}, OP_NOT_IN, "",         {"1", "2", "3"}},
+                      {"!=",  {"1"},           "!*=", {"2", "3"},      OP_NOT_IN, "",         {"1", "2", "3"}},
+                      {"!=",  {"3"},           "!*=", {"1", "2"},      OP_NOT_IN, "",         {"1", "2", "3"}},
+
+                      {"<",   {"1"},           "<",   {"1"},           OP_LT,     "1",        {}},
+                      {"<",   {"1"},           "<",   {"2"},           OP_LT,     "1",        {}},
+                      {"<",   {"2"},           "<",   {"1"},           OP_LT,     "1",        {}},
+                      {"<",   {"1"},           "<=",  {"1"},           OP_LT,     "1",        {}},
+
+                      {"<=",  {"1"},           "<=",  {"1"},           OP_LE,     "1",        {}},
+                      {"<=",  {"1"},           "<=",  {"2"},           OP_LE,     "1",        {}},
+                      {"<=",  {"2"},           "<=",  {"1"},           OP_LE,     "1",        {}},
+                      {"<=",  {"1"},           "<",   {"1"},           OP_LT,     "1",        {}},
+
+                      {">",   {"1"},           ">",   {"1"},           OP_GT,     "1",        {}},
+                      {">",   {"1"},           ">",   {"2"},           OP_GT,     "2",        {}},
+                      {">",   {"2"},           ">",   {"1"},           OP_GT,     "2",        {}},
+                      {">",   {"1"},           ">=",  {"1"},           OP_GT,     "1",        {}},
+
+                      {">=",  {"1"},           ">=",  {"1"},           OP_GE,     "1",        {}},
+                      {">=",  {"1"},           ">=",  {"2"},           OP_GE,     "2",        {}},
+                      {">=",  {"2"},           ">=",  {"1"},           OP_GE,     "2",        {}},
+                      {">=",  {"1"},           ">",   {"1"},           OP_GT,     "1",        {}},
+
+                      {"*=",  {"1"},           "*=",  {"2"},           OP_NULL,   "",         {}},
+                      {"*=",  {"1", "2", "3"}, "*=",  {"2"},           OP_EQ,     "2",        {}},
+                      {"*=",  {"1", "2", "3"}, "*=",  {"2", "3"},      OP_IN,     "",         {"2", "3"}},
+                      {"*=",  {"1"},           "=",   {"2"},           OP_NULL,   "",         {}},
+                      {"*=",  {"1", "2"},      "=",   {"1"},           OP_EQ,     "1",        {}},
+
+                      {"!*=", {"1", "2"},      "!*=", {"1", "3"},      OP_NOT_IN, "",         {"1", "2", "3"}},
+                      {"!*=", {"1", "2", "3"}, "!=",  {"2"},           OP_NOT_IN, "",         {"1", "2", "3"}},
+                      {"!*=", {"1", "2"},      "!=",  {"3"},           OP_NOT_IN, "",         {"1", "2", "3"}},
+
+                      {"is",  {"NULL"},        "is",  {"NOT NULL"},    OP_NULL,   "",         {}},
+                      {"is",  {"NOT NULL"},    "is",  {"NULL"},        OP_NULL,   "",         {}},
+                      {"is",  {"NULL"},        "is",  {"NULL"},        OP_IS,     "NULL",     {}},
+                      {"is",  {"NOT NULL"},    "is",  {"NOT NULL"},    OP_IS,     "NOT NULL", {}}};
+
+    int i = 0;
+    for (const auto &test : test_cases) {
+        ASSERT_NO_FATAL_FAILURE(calc_and_check(true,
+                                               create_cond(test.op1, test.operands1),
+                                               create_cond(test.op2, test.operands2),
+                                               test.result_op,
+                                               test.result_operand,
+                                               test.result_operands)) << "error index: " << i;
+        ++i;
+    }
+}
+
+TEST_F(CondTest, UnionCondsTest) {
+    struct {
+        std::string op1;
+        std::vector<std::string> operands1;
+        std::string op2;
+        std::vector<std::string> operands2;
+        CondOp result_op;
+        std::string result_operand;
+        std::set<std::string> result_operands;
+    } test_cases[] = {{"=",   {"1"},           "=",   {"1"},           OP_EQ,     "1",        {}},
+                      {"=",   {"1"},           "=",   {"2"},           OP_IN,     "",         {"1", "2"}},
+                      {"=",   {"1"},           "*=",  {"1", "2"},      OP_IN,     "",         {"1", "2"}},
+                      {"=",   {"1"},           "*=",  {"2", "3"},      OP_IN,     "",         {"1", "2", "3"}},
+
+                      {"!=",  {"1"},           "!=",  {"1"},           OP_NE,     "1",        {}},
+                      {"!=",  {"1"},           "!=",  {"2"},           OP_ALL,    "",         {}},
+                      {"!=",  {"1"},           "!*=", {"2", "3"},      OP_ALL,    "",         {}},
+                      {"!=",  {"1"},           "!*=", {"1", "2"},      OP_NE,     "1",        {}},
+
+                      {"<",   {"1"},           "<",   {"1"},           OP_LT,     "1",        {}},
+                      {"<",   {"1"},           "<",   {"2"},           OP_LT,     "2",        {}},
+                      {"<",   {"2"},           "<",   {"1"},           OP_LT,     "2",        {}},
+                      {"<",   {"1"},           "<=",  {"1"},           OP_LE,     "1",        {}},
+
+                      {"<=",  {"1"},           "<=",  {"1"},           OP_LE,     "1",        {}},
+                      {"<=",  {"1"},           "<=",  {"2"},           OP_LE,     "2",        {}},
+                      {"<=",  {"2"},           "<=",  {"1"},           OP_LE,     "2",        {}},
+                      {"<=",  {"1"},           "<",   {"1"},           OP_LE,     "1",        {}},
+
+                      {">",   {"1"},           ">",   {"1"},           OP_GT,     "1",        {}},
+                      {">",   {"1"},           ">",   {"2"},           OP_GT,     "1",        {}},
+                      {">",   {"2"},           ">",   {"1"},           OP_GT,     "1",        {}},
+                      {">",   {"1"},           ">=",  {"1"},           OP_GE,     "1",        {}},
+
+                      {">=",  {"1"},           ">=",  {"1"},           OP_GE,     "1",        {}},
+                      {">=",  {"1"},           ">=",  {"2"},           OP_GE,     "1",        {}},
+                      {">=",  {"2"},           ">=",  {"1"},           OP_GE,     "1",        {}},
+                      {">=",  {"1"},           ">",   {"1"},           OP_GE,     "1",        {}},
+
+                      {"*=",  {"1"},           "*=",  {"2"},           OP_IN,     "",         {"1", "2"}},
+                      {"*=",  {"1", "2", "3"}, "*=",  {"2", "4"},      OP_IN,     "",         {"1", "2", "3", "4"}},
+                      {"*=",  {"1", "2", "3"}, "=",   {"2"},           OP_IN,     "",         {"1", "2", "3"}},
+                      {"*=",  {"2", "3"},      "=",   {"1"},           OP_IN,     "",         {"1", "2", "3"}},
+                      {"*=",  {"1", "2"},      "=",   {"3"},           OP_IN,     "",         {"1", "2", "3"}},
+
+                      {"!*=", {"1"},           "!*=", {"2"},           OP_ALL,    "",         {}},
+                      {"!*=", {"1", "2", "3"}, "!*=", {"2", "4"},      OP_NE,     "2",        {}},
+                      {"!*=", {"1", "2", "3"}, "!*=", {"2", "3", "4"}, OP_NOT_IN, "",         {"2", "3"}},
+                      {"!*=", {"1", "2"},      "!=",  {"3"},           OP_ALL,    "",         {}},
+                      {"!*=", {"1", "2"},      "!=",  {"1"},           OP_NE,     "1",        {}},
+
+                      {"is",  {"NULL"},        "is",  {"NOT NULL"},    OP_ALL,    "",         {}},
+                      {"is",  {"NOT NULL"},    "is",  {"NULL"},        OP_ALL,    "",         {}},
+                      {"is",  {"NULL"},        "is",  {"NULL"},        OP_IS,     "NULL",     {}},
+                      {"is",  {"NOT NULL"},    "is",  {"NOT NULL"},    OP_IS,     "NOT NULL", {}}};
+    int i = 0;
+    for (const auto &test : test_cases) {
+        ASSERT_NO_FATAL_FAILURE(calc_and_check(false,
+                                               create_cond(test.op1, test.operands1),
+                                               create_cond(test.op2, test.operands2),
+                                               test.result_op,
+                                               test.result_operand,
+                                               test.result_operands)) << "error index: " << i;
+        ++i;
+    }
+}
+
+}  // namespace doris
+
+int main(int argc, char** argv) {
+    testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}
diff --git a/be/test/olap/rowset/segment_v2/segment_test.cpp b/be/test/olap/rowset/segment_v2/segment_test.cpp
index f4971b6ce269c8..1673fd292f2ba0 100644
--- a/be/test/olap/rowset/segment_v2/segment_test.cpp
+++ b/be/test/olap/rowset/segment_v2/segment_test.cpp
@@ -444,6 +444,7 @@ TEST_F(SegmentReaderWriterTest, TestIndex) {
             std::shared_ptr<Conditions> conditions(new Conditions());
             conditions->set_tablet_schema(&tablet_schema);
             ASSERT_EQ(OLAP_SUCCESS, conditions->append_condition(condition));
+            conditions->normalize();
 
             StorageReadOptions read_opts;
             read_opts.stats = &stats;
@@ -467,6 +468,7 @@ TEST_F(SegmentReaderWriterTest, TestIndex) {
             std::shared_ptr<Conditions> conditions(new Conditions());
             conditions->set_tablet_schema(&tablet_schema);
             ASSERT_EQ(OLAP_SUCCESS, conditions->append_condition(condition));
+            conditions->normalize();
 
             StorageReadOptions read_opts;
             read_opts.stats = &stats;
@@ -516,6 +518,7 @@ TEST_F(SegmentReaderWriterTest, TestIndex) {
             std::shared_ptr<Conditions> conditions(new Conditions());
             conditions->set_tablet_schema(&tablet_schema);
             ASSERT_EQ(OLAP_SUCCESS, conditions->append_condition(condition));
+            conditions->normalize();
 
             // the second page read will be pruned by the following delete predicate
             TCondition delete_condition;
@@ -526,6 +529,7 @@ TEST_F(SegmentReaderWriterTest, TestIndex) {
             std::shared_ptr<Conditions> delete_conditions(new Conditions());
             delete_conditions->set_tablet_schema(&tablet_schema);
             ASSERT_EQ(OLAP_SUCCESS, delete_conditions->append_condition(delete_condition));
+            delete_conditions->normalize();
 
             StorageReadOptions read_opts;
             read_opts.stats = &stats;
@@ -579,6 +583,7 @@ TEST_F(SegmentReaderWriterTest, TestIndex) {
             std::shared_ptr<Conditions> conditions(new Conditions());
             conditions->set_tablet_schema(&tablet_schema);
             ASSERT_EQ(OLAP_SUCCESS, conditions->append_condition(condition));
+            conditions->normalize();
             read_opts.conditions = conditions.get();
             std::unique_ptr<RowwiseIterator> iter;
             segment->new_iterator(schema, read_opts, &iter);
@@ -931,6 +936,7 @@ TEST_F(SegmentReaderWriterTest, TestStringDict) {
             std::shared_ptr<Conditions> conditions(new Conditions());
             conditions->set_tablet_schema(tablet_schema.get());
             ASSERT_EQ(OLAP_SUCCESS, conditions->append_condition(condition));
+            conditions->normalize();
 
             StorageReadOptions read_opts;
             read_opts.stats = &stats;
@@ -988,6 +994,7 @@ TEST_F(SegmentReaderWriterTest, TestStringDict) {
             std::shared_ptr<Conditions> conditions(new Conditions());
             conditions->set_tablet_schema(tablet_schema.get());
             ASSERT_EQ(OLAP_SUCCESS, conditions->append_condition(condition));
+            conditions->normalize();
 
             StorageReadOptions read_opts;
             read_opts.stats = &stats;
diff --git a/build.sh b/build.sh
index 8f5215019aff3d..7cf7f1dd5aed0f 100755
--- a/build.sh
+++ b/build.sh
@@ -41,7 +41,7 @@ if [[ ! -f ${DORIS_THIRDPARTY}/installed/lib/libs2.a ]]; then
     ${DORIS_THIRDPARTY}/build-thirdparty.sh
 fi
 
-PARALLEL=$[$(nproc)/4+1]
+PARALLEL=32 #$[$(nproc)/4+1]
 
 # Check args
 usage() {
diff --git a/run-be-ut.sh b/run-be-ut.sh
index f9c205efa85c0d..9727b4372e0c8e 100755
--- a/run-be-ut.sh
+++ b/run-be-ut.sh
@@ -39,7 +39,7 @@ export DORIS_HOME=${ROOT}
 
 . ${DORIS_HOME}/env.sh
 
-PARALLEL=$[$(nproc)/4+1]
+PARALLEL=32 #$[$(nproc)/4+1]
 
 # Check args
 usage() {