diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
index 1de34109c5..b0940c45f1 100644
--- a/be/src/olap/compaction.cpp
+++ b/be/src/olap/compaction.cpp
@@ -385,6 +385,7 @@ Status Compaction::do_compaction_impl(int64_t permits) {
                 });
         // now version in delete_predicate is deprecated
         if (!delete_predicate.in_predicates().empty() ||
+            !delete_predicate.sub_predicates_v2().empty() ||
             !delete_predicate.sub_predicates().empty()) {
             _output_rowset->rowset_meta()->set_delete_predicate(std::move(delete_predicate));
         }
diff --git a/be/src/olap/data_dir.cpp b/be/src/olap/data_dir.cpp
index 9fa6f7ccc9..bc488d06a4 100644
--- a/be/src/olap/data_dir.cpp
+++ b/be/src/olap/data_dir.cpp
@@ -17,6 +17,7 @@
 
 #include "olap/data_dir.h"
 
+#include <fmt/core.h>
 #include <fmt/format.h>
 #include <gen_cpp/Types_types.h>
 #include <gen_cpp/olap_file.pb.h>
@@ -25,6 +26,7 @@
 #include <atomic>
 // IWYU pragma: no_include <bits/chrono.h>
 #include <chrono> // IWYU pragma: keep
+#include <cstddef>
 #include <filesystem>
 #include <memory>
 #include <new>
@@ -43,6 +45,7 @@
 #include "io/fs/local_file_system.h"
 #include "io/fs/path.h"
 #include "io/fs/remote_file_system.h"
+#include "olap/delete_handler.h"
 #include "olap/olap_common.h"
 #include "olap/olap_define.h"
 #include "olap/olap_meta.h"
@@ -366,7 +369,7 @@ Status DataDir::load() {
 
     std::vector<RowsetMetaSharedPtr> dir_rowset_metas;
     LOG(INFO) << "begin loading rowset from meta";
-    auto load_rowset_func = [&dir_rowset_metas, &local_fs = fs()](
+    auto load_rowset_func = [&dir_rowset_metas, &local_fs = fs(), this](
                                     TabletUid tablet_uid, RowsetId rowset_id,
                                     const std::string& meta_str) -> bool {
         RowsetMetaSharedPtr rowset_meta(new RowsetMeta());
@@ -379,6 +382,34 @@ Status DataDir::load() {
         if (rowset_meta->is_local()) {
             rowset_meta->set_fs(local_fs);
         }
+        if (rowset_meta->has_delete_predicate()) {
+            // copy the delete sub pred v1 to check then
+            auto orig_delete_sub_pred = rowset_meta->delete_predicate().sub_predicates();
+            auto* delete_pred = rowset_meta->mutable_delete_pred_pb();
+
+            if ((!delete_pred->sub_predicates().empty() &&
+                 delete_pred->sub_predicates_v2().empty()) ||
+                (!delete_pred->in_predicates().empty() &&
+                 delete_pred->in_predicates()[0].has_column_unique_id())) {
+                // convert pred and write only when delete sub pred v2 is not set or there is in list pred to be set column uid
+                DeleteHandler::convert_to_sub_pred_v2(delete_pred, rowset_meta->tablet_schema());
+                LOG(INFO) << fmt::format(
+                        "convert rowset with old delete pred: rowset_id={}, tablet_id={}",
+                        rowset_id.to_string(), tablet_uid.to_string());
+                CHECK_EQ(orig_delete_sub_pred.size(), delete_pred->sub_predicates().size())
+                        << "inconsistent sub predicate v1 after conversion";
+                for (size_t i = 0; i < orig_delete_sub_pred.size(); ++i) {
+                    CHECK_STREQ(orig_delete_sub_pred.Get(i).c_str(),
+                                delete_pred->sub_predicates().Get(i).c_str())
+                            << "inconsistent sub predicate v1 after conversion";
+                }
+                std::string result;
+                rowset_meta->serialize(&result);
+                std::string key =
+                        ROWSET_PREFIX + tablet_uid.to_string() + "_" + rowset_id.to_string();
+                _meta->put(META_COLUMN_FAMILY_INDEX, key, result);
+            }
+        }
         dir_rowset_metas.push_back(rowset_meta);
         return true;
     };
diff --git a/be/src/olap/delete_handler.cpp b/be/src/olap/delete_handler.cpp
index 46bfe56cb7..df0f6211aa 100644
--- a/be/src/olap/delete_handler.cpp
+++ b/be/src/olap/delete_handler.cpp
@@ -35,6 +35,7 @@
 #include "olap/column_predicate.h"
 #include "olap/olap_common.h"
 #include "olap/predicate_creator.h"
+#include "olap/tablet_schema.h"
 #include "olap/utils.h"
 
 using apache::thrift::ThriftDebugString;
@@ -47,6 +48,8 @@ using std::regex_error;
 using std::regex_match;
 using std::smatch;
 
+using ::google::protobuf::RepeatedPtrField;
+
 namespace doris {
 using namespace ErrorCode;
 
@@ -71,6 +74,9 @@ Status DeleteHandler::generate_delete_predicate(const TabletSchema& schema,
     for (const TCondition& condition : conditions) {
         if (condition.condition_values.size() > 1) {
             InPredicatePB* in_pred = del_pred->add_in_predicates();
+            if (condition.__isset.column_unique_id) {
+                in_pred->set_column_unique_id(condition.column_unique_id);
+            }
             in_pred->set_column_name(condition.column_name);
             bool is_not_in = condition.condition_op == "!*=";
             in_pred->set_is_not_in(is_not_in);
@@ -81,9 +87,18 @@ Status DeleteHandler::generate_delete_predicate(const TabletSchema& schema,
             LOG(INFO) << "store one sub-delete condition. condition name=" << in_pred->column_name()
                       << "condition size=" << in_pred->values().size();
         } else {
-            string condition_str = construct_sub_predicates(condition);
-            del_pred->add_sub_predicates(condition_str);
-            LOG(INFO) << "store one sub-delete condition. condition=" << condition_str;
+            // write sub predicate v1 for compactbility
+            del_pred->add_sub_predicates(construct_sub_predicate(condition));
+            DeleteSubPredicatePB* sub_predicate = del_pred->add_sub_predicates_v2();
+            if (condition.__isset.column_unique_id) {
+                sub_predicate->set_column_unique_id(condition.column_unique_id);
+            }
+            sub_predicate->set_column_name(condition.column_name);
+            sub_predicate->set_op(trans_op(condition.condition_op));
+            sub_predicate->set_cond_value(condition.condition_values[0]);
+            LOG(INFO) << "store one sub-delete condition. condition="
+                      << fmt::format(" {} {} {}", condition.column_name, condition.condition_op,
+                                     condition.condition_values[0]);
         }
     }
     del_pred->set_version(-1);
@@ -91,7 +106,27 @@ Status DeleteHandler::generate_delete_predicate(const TabletSchema& schema,
     return Status::OK();
 }
 
-std::string DeleteHandler::construct_sub_predicates(const TCondition& condition) {
+void DeleteHandler::convert_to_sub_pred_v2(DeletePredicatePB* delete_pred,
+                                           TabletSchemaSPtr schema) {
+    if (!delete_pred->sub_predicates().empty() && delete_pred->sub_predicates_v2().empty()) {
+        for (const auto& condition_str : delete_pred->sub_predicates()) {
+            auto* sub_pred = delete_pred->add_sub_predicates_v2();
+            TCondition condition;
+            parse_condition(condition_str, &condition);
+            sub_pred->set_column_unique_id(schema->column(condition.column_name).unique_id());
+            sub_pred->set_column_name(condition.column_name);
+            sub_pred->set_op(condition.condition_op);
+            sub_pred->set_cond_value(condition.condition_values[0]);
+        }
+    }
+
+    auto* in_pred_list = delete_pred->mutable_in_predicates();
+    for (auto& in_pred : *in_pred_list) {
+        in_pred.set_column_unique_id(schema->column(in_pred.column_name()).unique_id());
+    }
+}
+
+std::string DeleteHandler::construct_sub_predicate(const TCondition& condition) {
     string op = condition.condition_op;
     if (op == "<") {
         op += "<";
@@ -112,6 +147,23 @@ std::string DeleteHandler::construct_sub_predicates(const TCondition& condition)
     return condition_str;
 }
 
+std::string DeleteHandler::trans_op(const std::string& opt) {
+    std::string op = string(opt);
+    if (op == "<") {
+        op += "<";
+    } else if (op == ">") {
+        op += ">";
+    }
+    if ("IS" != op) {
+        if (op == "*=") {
+            op = "=";
+        } else if (op == "!*=") {
+            op = "!=";
+        }
+    }
+    return op;
+}
+
 bool DeleteHandler::is_condition_value_valid(const TabletColumn& column,
                                              const std::string& condition_op,
                                              const string& value_str) {
@@ -200,7 +252,22 @@ Status DeleteHandler::check_condition_valid(const TabletSchema& schema, const TC
     return Status::OK();
 }
 
-bool DeleteHandler::_parse_condition(const std::string& condition_str, TCondition* condition) {
+Status DeleteHandler::parse_condition(const DeleteSubPredicatePB& sub_cond, TCondition* condition) {
+    if (!sub_cond.has_column_name() || !sub_cond.has_op() || !sub_cond.has_cond_value()) {
+        return Status::Error<DELETE_INVALID_PARAMETERS>(
+                "fail to parse condition. condition={} {} {}", sub_cond.column_name(),
+                sub_cond.op(), sub_cond.cond_value());
+    }
+    if (sub_cond.has_column_unique_id()) {
+        condition->column_unique_id = sub_cond.column_unique_id();
+    }
+    condition->column_name = sub_cond.column_name();
+    condition->condition_op = sub_cond.op();
+    condition->condition_values.push_back(sub_cond.cond_value());
+    return Status::OK();
+}
+
+Status DeleteHandler::parse_condition(const std::string& condition_str, TCondition* condition) {
     bool matched = true;
     smatch what;
 
@@ -227,7 +294,8 @@ bool DeleteHandler::_parse_condition(const std::string& condition_str, TConditio
     }
 
     if (!matched) {
-        return false;
+        return Status::Error<DELETE_INVALID_PARAMETERS>("fail to sub condition. condition={}",
+                                                        condition_str);
     }
     condition->column_name = what[1].str();
     condition->condition_op = what[2].str();
@@ -237,11 +305,44 @@ bool DeleteHandler::_parse_condition(const std::string& condition_str, TConditio
         condition->condition_values.push_back(what[3].str());
     }
 
-    return true;
+    return Status::OK();
 }
 
+template <typename SubPredType>
+Status DeleteHandler::_parse_column_pred(TabletSchemaSPtr complete_schema,
+                                         TabletSchemaSPtr delete_pred_related_schema,
+                                         const RepeatedPtrField<SubPredType>& sub_pred_list,
+                                         DeleteConditions* delete_conditions) {
+    for (const auto& sub_predicate : sub_pred_list) {
+        TCondition condition;
+        RETURN_IF_ERROR(parse_condition(sub_predicate, &condition));
+        int32_t col_unique_id =
+                delete_pred_related_schema->column(condition.column_name).unique_id();
+        condition.__set_column_unique_id(col_unique_id);
+        const auto& column = complete_schema->column_by_uid(col_unique_id);
+        uint32_t index = complete_schema->field_index(col_unique_id);
+        auto* predicate =
+                parse_to_predicate(column, index, condition, _predicate_arena.get(), true);
+        if (predicate != nullptr) {
+            delete_conditions->column_predicate_vec.push_back(predicate);
+        }
+    }
+    return Status::OK();
+}
+
+template Status DeleteHandler::_parse_column_pred<DeleteSubPredicatePB>(
+        TabletSchemaSPtr complete_schema, TabletSchemaSPtr delete_pred_related_schema,
+        const ::google::protobuf::RepeatedPtrField<DeleteSubPredicatePB>& sub_pred_list,
+        DeleteConditions* delete_conditions);
+
+template Status DeleteHandler::_parse_column_pred<std::string>(
+        TabletSchemaSPtr complete_schema, TabletSchemaSPtr delete_pred_related_schema,
+        const ::google::protobuf::RepeatedPtrField<std::string>& sub_pred_list,
+        DeleteConditions* delete_conditions);
+
 Status DeleteHandler::init(TabletSchemaSPtr tablet_schema,
-                           const std::vector<RowsetMetaSharedPtr>& delete_preds, int64_t version) {
+                           const std::vector<RowsetMetaSharedPtr>& delete_preds, int64_t version,
+                           bool with_sub_pred_v2) {
     DCHECK(!_is_inited) << "reinitialize delete handler.";
     DCHECK(version >= 0) << "invalid parameters. version=" << version;
     _predicate_arena.reset(new vectorized::Arena());
@@ -256,26 +357,20 @@ Status DeleteHandler::init(TabletSchemaSPtr tablet_schema,
         auto& delete_condition = delete_pred->delete_predicate();
         DeleteConditions temp;
         temp.filter_version = delete_pred->version().first;
-        for (const auto& sub_predicate : delete_condition.sub_predicates()) {
-            TCondition condition;
-            if (!_parse_condition(sub_predicate, &condition)) {
-                return Status::Error<DELETE_INVALID_PARAMETERS>(
-                        "fail to parse condition. condition={}", sub_predicate);
-            }
-            int32_t col_unique_id =
-                    delete_pred_related_schema->column(condition.column_name).unique_id();
-            const auto& column = tablet_schema->column_by_uid(col_unique_id);
-            uint32_t index = tablet_schema->field_index(col_unique_id);
-            auto predicate =
-                    parse_to_predicate(column, index, condition, _predicate_arena.get(), true);
-            if (predicate != nullptr) {
-                temp.column_predicate_vec.push_back(predicate);
-            }
+        if (with_sub_pred_v2) {
+            RETURN_IF_ERROR(_parse_column_pred(tablet_schema, delete_pred_related_schema,
+                                               delete_condition.sub_predicates_v2(), &temp));
+        } else {
+            // make it compatible with the former versions
+            RETURN_IF_ERROR(_parse_column_pred(tablet_schema, delete_pred_related_schema,
+                                               delete_condition.sub_predicates(), &temp));
         }
-
         for (const auto& in_predicate : delete_condition.in_predicates()) {
             TCondition condition;
             condition.__set_column_name(in_predicate.column_name());
+            condition.__set_column_unique_id(
+                    delete_pred_related_schema->column(condition.column_name).unique_id());
+
             if (in_predicate.is_not_in()) {
                 condition.__set_condition_op("!*=");
             } else {
diff --git a/be/src/olap/delete_handler.h b/be/src/olap/delete_handler.h
index 7239f4326b..429de58b4e 100644
--- a/be/src/olap/delete_handler.h
+++ b/be/src/olap/delete_handler.h
@@ -62,8 +62,7 @@ public:
                                             const std::vector<TCondition>& conditions,
                                             DeletePredicatePB* del_pred);
 
-    // construct sub condition from TCondition
-    static std::string construct_sub_predicates(const TCondition& condition);
+    static void convert_to_sub_pred_v2(DeletePredicatePB* delete_pred, TabletSchemaSPtr schema);
 
 private:
     // Validate the condition on the schema.
@@ -78,6 +77,16 @@ private:
                                          const std::string& condition_op,
                                          const std::string& value_str);
 
+    // construct sub condition from TCondition
+    static std::string construct_sub_predicate(const TCondition& condition);
+
+    // make operators from FE adaptive to BE
+    [[nodiscard]] static std::string trans_op(const string& op);
+
+    // extract 'column_name', 'op' and 'operands' to condition
+    static Status parse_condition(const DeleteSubPredicatePB& sub_cond, TCondition* condition);
+    static Status parse_condition(const std::string& condition_str, TCondition* condition);
+
 public:
     DeleteHandler() = default;
     ~DeleteHandler() { finalize(); }
@@ -85,17 +94,18 @@ public:
     // Initialize DeleteHandler, use the delete conditions of this tablet whose version less than or equal to
     // 'version' to fill '_del_conds'.
     // NOTE: You should lock the tablet's header file before calling this function.
-    //
     // input:
     //     * schema: tablet's schema, the delete conditions and data rows are in this schema
     //     * version: maximum version
+    //     * with_sub_pred_v2: whether to use delete sub predicate v2 (v2 is based on PB, v1 is based on condition string)
     // return:
     //     * Status::Error<DELETE_INVALID_PARAMETERS>(): input parameters are not valid
     //     * Status::Error<MEM_ALLOC_FAILED>(): alloc memory failed
     Status init(TabletSchemaSPtr tablet_schema,
-                const std::vector<RowsetMetaSharedPtr>& delete_conditions, int64_t version);
+                const std::vector<RowsetMetaSharedPtr>& delete_conditions, int64_t version,
+                bool with_sub_pred_v2 = false);
 
-    bool empty() const { return _del_conds.empty(); }
+    [[nodiscard]] bool empty() const { return _del_conds.empty(); }
 
     // Release an instance of this class.
     void finalize();
@@ -106,8 +116,11 @@ public:
                     del_predicates_for_zone_map) const;
 
 private:
-    // Use regular expression to extract 'column_name', 'op' and 'operands'
-    bool _parse_condition(const std::string& condition_str, TCondition* condition);
+    template <typename SubPredType>
+    Status _parse_column_pred(
+            TabletSchemaSPtr complete_schema, TabletSchemaSPtr delete_pred_related_schema,
+            const ::google::protobuf::RepeatedPtrField<SubPredType>& sub_pred_list,
+            DeleteConditions* delete_conditions);
 
     bool _is_inited = false;
     // DeleteConditions in _del_conds are in 'OR' relationship
diff --git a/be/src/olap/olap_meta.h b/be/src/olap/olap_meta.h
index 504a07f90f..7432c883b6 100644
--- a/be/src/olap/olap_meta.h
+++ b/be/src/olap/olap_meta.h
@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include <rocksdb/iterator.h>
+
 #include <functional>
 #include <memory>
 #include <string>
@@ -60,17 +62,21 @@ public:
 
     Status iterate(const int column_family_index, const std::string& prefix,
                    std::function<bool(const std::string&, const std::string&)> const& func);
+
     Status iterate(const int column_family_index, const std::string& seek_key,
                    const std::string& prefix,
                    std::function<bool(const std::string&, const std::string&)> const& func);
 
-    std::string get_root_path() const { return _root_path; }
+    [[nodiscard]] std::string get_root_path() const { return _root_path; }
 
     rocksdb::ColumnFamilyHandle* get_handle(const int column_family_index) {
         return _handles[column_family_index].get();
     }
 
 private:
+    Status get_iterator(const int column_family_index, const std::string& seek_key,
+                        const std::string& prefix, rocksdb::Iterator*);
+
     std::string _root_path;
     // keep order of _db && _handles, we need destroy _handles before _db
     std::unique_ptr<rocksdb::DB, std::function<void(rocksdb::DB*)>> _db;
diff --git a/be/src/olap/reader.cpp b/be/src/olap/reader.cpp
index b86217d537..040f98e7e2 100644
--- a/be/src/olap/reader.cpp
+++ b/be/src/olap/reader.cpp
@@ -615,9 +615,17 @@ Status TabletReader::_init_delete_condition(const ReaderParams& read_params) {
                       ((read_params.reader_type == ReaderType::READER_CUMULATIVE_COMPACTION &&
                         config::enable_delete_when_cumu_compaction)) ||
                       read_params.reader_type == ReaderType::READER_CHECKSUM);
-
+    if (_filter_delete) {
+        // note(tsy): for compaction, keep delete sub pred v1 temporarily
+        return _delete_handler.init(_tablet_schema, read_params.delete_predicates,
+                                    read_params.version.second, false);
+    }
+    auto* runtime_state = read_params.runtime_state;
+    // note(tsy): for query, use session var to enable delete sub pred v2, for schema change, use v2 directly
+    bool enable_sub_pred_v2 =
+            runtime_state == nullptr ? true : runtime_state->enable_delete_sub_pred_v2();
     return _delete_handler.init(_tablet_schema, read_params.delete_predicates,
-                                read_params.version.second);
+                                read_params.version.second, enable_sub_pred_v2);
 }
 
 Status TabletReader::init_reader_params_and_create_block(
diff --git a/be/src/olap/rowset/rowset_meta.h b/be/src/olap/rowset/rowset_meta.h
index 360a3a0202..d4216b5a01 100644
--- a/be/src/olap/rowset/rowset_meta.h
+++ b/be/src/olap/rowset/rowset_meta.h
@@ -265,6 +265,10 @@ public:
         return rowset_meta_pb;
     }
 
+    inline DeletePredicatePB* mutable_delete_pred_pb() {
+        return _rowset_meta_pb.mutable_delete_predicate();
+    }
+
     bool is_singleton_delta() const {
         return has_version() && _rowset_meta_pb.start_version() == _rowset_meta_pb.end_version();
     }
diff --git a/be/src/olap/rowset/rowset_meta_manager.cpp b/be/src/olap/rowset/rowset_meta_manager.cpp
index 5a0f968c72..2315d76c98 100644
--- a/be/src/olap/rowset/rowset_meta_manager.cpp
+++ b/be/src/olap/rowset/rowset_meta_manager.cpp
@@ -36,9 +36,6 @@
 #include "olap/utils.h"
 
 namespace doris {
-namespace {
-const std::string ROWSET_PREFIX = "rst_";
-} // namespace
 
 using namespace ErrorCode;
 
@@ -414,8 +411,7 @@ Status RowsetMetaManager::remove(OlapMeta* meta, TabletUid tablet_uid, const Row
 
 Status RowsetMetaManager::remove_binlog(OlapMeta* meta, const std::string& suffix) {
     return meta->remove(META_COLUMN_FAMILY_INDEX,
-                        std::vector<std::string> {kBinlogMetaPrefix.data() + suffix,
-                                                  kBinlogDataPrefix.data() + suffix});
+                        {kBinlogMetaPrefix.data() + suffix, kBinlogDataPrefix.data() + suffix});
 }
 
 Status RowsetMetaManager::ingest_binlog_metas(OlapMeta* meta, TabletUid tablet_uid,
diff --git a/be/src/olap/rowset/rowset_meta_manager.h b/be/src/olap/rowset/rowset_meta_manager.h
index 0c04cb686c..6d7f092040 100644
--- a/be/src/olap/rowset/rowset_meta_manager.h
+++ b/be/src/olap/rowset/rowset_meta_manager.h
@@ -35,6 +35,9 @@ class RowsetMetaPB;
 } // namespace doris
 
 namespace doris {
+namespace {
+const std::string ROWSET_PREFIX = "rst_";
+} // namespace
 
 // Helper class for managing rowset meta of one root path.
 class RowsetMetaManager {
diff --git a/be/src/runtime/runtime_state.h b/be/src/runtime/runtime_state.h
index add158dafa..08245b742c 100644
--- a/be/src/runtime/runtime_state.h
+++ b/be/src/runtime/runtime_state.h
@@ -434,6 +434,11 @@ public:
                        : 0;
     }
 
+    inline bool enable_delete_sub_pred_v2() const {
+        return _query_options.__isset.enable_delete_sub_predicate_v2 &&
+               _query_options.enable_delete_sub_predicate_v2;
+    }
+
     void emplace_local_state(int id, std::shared_ptr<doris::pipeline::PipelineXLocalState> state);
 
     std::shared_ptr<doris::pipeline::PipelineXLocalState> get_local_state(int id);
diff --git a/be/test/olap/delete_handler_test.cpp b/be/test/olap/delete_handler_test.cpp
index c2aa5f1efe..52b2c1873b 100644
--- a/be/test/olap/delete_handler_test.cpp
+++ b/be/test/olap/delete_handler_test.cpp
@@ -393,6 +393,33 @@ TEST_F(TestDeleteConditionHandler, StoreCondSucceed) {
     EXPECT_STREQ("k12!='9'", del_pred.sub_predicates(5).c_str());
     EXPECT_STREQ("k$1>>'1'", del_pred.sub_predicates(6).c_str());
 
+    // check sub predicate v2
+
+    EXPECT_EQ(size_t(7), del_pred.sub_predicates_v2_size());
+    EXPECT_STREQ("k1", del_pred.sub_predicates_v2(0).column_name().c_str());
+    EXPECT_STREQ("k2", del_pred.sub_predicates_v2(1).column_name().c_str());
+    EXPECT_STREQ("k3", del_pred.sub_predicates_v2(2).column_name().c_str());
+    EXPECT_STREQ("k4", del_pred.sub_predicates_v2(3).column_name().c_str());
+    EXPECT_STREQ("k5", del_pred.sub_predicates_v2(4).column_name().c_str());
+    EXPECT_STREQ("k12", del_pred.sub_predicates_v2(5).column_name().c_str());
+    EXPECT_STREQ("k$1", del_pred.sub_predicates_v2(6).column_name().c_str());
+
+    EXPECT_STREQ("=", del_pred.sub_predicates_v2(0).op().c_str());
+    EXPECT_STREQ(">>", del_pred.sub_predicates_v2(1).op().c_str());
+    EXPECT_STREQ("<=", del_pred.sub_predicates_v2(2).op().c_str());
+    EXPECT_STREQ("IS", del_pred.sub_predicates_v2(3).op().c_str());
+    EXPECT_STREQ("=", del_pred.sub_predicates_v2(4).op().c_str());
+    EXPECT_STREQ("!=", del_pred.sub_predicates_v2(5).op().c_str());
+    EXPECT_STREQ(">>", del_pred.sub_predicates_v2(6).op().c_str());
+
+    EXPECT_STREQ("1", del_pred.sub_predicates_v2(0).cond_value().c_str());
+    EXPECT_STREQ("3", del_pred.sub_predicates_v2(1).cond_value().c_str());
+    EXPECT_STREQ("5", del_pred.sub_predicates_v2(2).cond_value().c_str());
+    EXPECT_STREQ("NULL", del_pred.sub_predicates_v2(3).cond_value().c_str());
+    EXPECT_STREQ("7", del_pred.sub_predicates_v2(4).cond_value().c_str());
+    EXPECT_STREQ("9", del_pred.sub_predicates_v2(5).cond_value().c_str());
+    EXPECT_STREQ("1", del_pred.sub_predicates_v2(6).cond_value().c_str());
+
     EXPECT_EQ(size_t(1), del_pred.in_predicates_size());
     EXPECT_FALSE(del_pred.in_predicates(0).is_not_in());
     EXPECT_STREQ("k13", del_pred.in_predicates(0).column_name().c_str());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index ee66fa3eb6..7e5c06d87e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -381,6 +381,8 @@ public class SessionVariable implements Serializable, Writable {
 
     public static final String ROUND_PRECISE_DECIMALV2_VALUE = "round_precise_decimalv2_value";
 
+    public static final String ENABLE_DELETE_SUB_PREDICATE_V2 = "enable_delete_sub_predicate_v2";
+
     public static final String JDBC_CLICKHOUSE_QUERY_FINAL = "jdbc_clickhouse_query_final";
 
     public static final String ENABLE_MEMTABLE_ON_SINK_NODE =
@@ -1109,6 +1111,9 @@ public class SessionVariable implements Serializable, Writable {
     @VariableMgr.VarAttr(name = PARALLEL_SYNC_ANALYZE_TASK_NUM)
     public int parallelSyncAnalyzeTaskNum = 2;
 
+    @VariableMgr.VarAttr(name = ENABLE_DELETE_SUB_PREDICATE_V2, fuzzy = true, needForward = true)
+    public boolean enableDeleteSubPredicateV2 = true;
+
     @VariableMgr.VarAttr(name = TRUNCATE_CHAR_OR_VARCHAR_COLUMNS,
             description = {"是否按照表的 schema 来截断 char 或者 varchar 列。默认为 false。\n"
                     + "因为外表会存在表的 schema 中 char 或者 varchar 列的最大长度和底层 parquet 或者 orc 文件中的 schema 不一致"
@@ -1144,9 +1149,11 @@ public class SessionVariable implements Serializable, Writable {
         if (randomInt % 2 == 0) {
             this.rewriteOrToInPredicateThreshold = 100000;
             this.enableFunctionPushdown = false;
+            this.enableDeleteSubPredicateV2 = false;
         } else {
             this.rewriteOrToInPredicateThreshold = 2;
             this.enableFunctionPushdown = true;
+            this.enableDeleteSubPredicateV2 = true;
         }
         this.runtimeFilterType = 1 << randomInt;
         /*
@@ -2205,6 +2212,7 @@ public class SessionVariable implements Serializable, Writable {
         tResult.setEnableParquetLazyMat(enableParquetLazyMat);
         tResult.setEnableOrcLazyMat(enableOrcLazyMat);
 
+        tResult.setEnableDeleteSubPredicateV2(enableDeleteSubPredicateV2);
         tResult.setTruncateCharOrVarcharColumns(truncateCharOrVarcharColumns);
         tResult.setEnableMemtableOnSinkNode(enableMemtableOnSinkNode);
 
diff --git a/gensrc/proto/olap_file.proto b/gensrc/proto/olap_file.proto
index fe3ac7915c..6a21a77734 100644
--- a/gensrc/proto/olap_file.proto
+++ b/gensrc/proto/olap_file.proto
@@ -149,12 +149,21 @@ message DeletePredicatePB {
     required int32 version = 1; // This field is useless, but could not removed, not depend on it
     repeated string sub_predicates = 2;
     repeated InPredicatePB in_predicates = 3;
+    repeated DeleteSubPredicatePB sub_predicates_v2 = 4;
+}
+
+message DeleteSubPredicatePB {
+    optional int32 column_unique_id = 1;
+    optional string column_name = 2;
+    optional string op = 3;
+    optional string cond_value = 4;
 }
 
 message InPredicatePB {
    optional string column_name = 1;
    optional bool is_not_in = 2;
    repeated string values = 3;
+   optional int32 column_unique_id = 4;
 }
 
 enum AlterTabletState {
diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift
index ba204f474e..a87238dfff 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -235,6 +235,8 @@ struct TQueryOptions {
   79: optional bool enable_pipeline_x_engine = false;
 
   80: optional bool enable_memtable_on_sink_node = false;
+
+  81: optional bool enable_delete_sub_predicate_v2 = false;
 }