diff --git a/be/src/exprs/function/like.cpp b/be/src/exprs/function/like.cpp
index 8bbfcf9b81dada..938ecfb958fa1e 100644
--- a/be/src/exprs/function/like.cpp
+++ b/be/src/exprs/function/like.cpp
@@ -19,7 +19,6 @@
 
 #include <fmt/format.h>
 #include <hs/hs_compile.h>
-#include <re2/stringpiece.h>
 
 #include <cstddef>
 #include <ostream>
@@ -34,8 +33,10 @@
 #include "core/column/column_vector.h"
 #include "core/string_ref.h"
 #include "exprs/function/simple_function_factory.h"
+#include "util/hyperscan_util.h"
 
 namespace doris {
+
 // A regex to match any regex pattern is equivalent to a substring search.
 static const RE2 SUBSTRING_RE(R"((?:\.\*)*([^\.\^\{\[\(\|\)\]\}\+\*\?\$\\]*)(?:\.\*)*)");
 
@@ -183,8 +184,9 @@ struct VectorEndsWithSearchState : public VectorPatternSearchState {
     }
 };
 
-Status LikeSearchState::clone(LikeSearchState& cloned) {
+Status LikeSearchState::clone(LikeSearchState& cloned) const {
     cloned.set_search_string(search_string);
+    cloned.enable_hyperscan_fallback = enable_hyperscan_fallback;
 
     std::string re_pattern;
     FunctionLike::convert_like_pattern(this, pattern_str, &re_pattern);
@@ -452,7 +454,8 @@ Status FunctionLikeBase::regexp_fn(const LikeSearchState* state, const ColumnStr
 
     hs_database_t* database = nullptr;
     hs_scratch_t* scratch = nullptr;
-    if (hs_prepare(nullptr, re_pattern.c_str(), &database, &scratch).ok()) { // use hyperscan
+    auto hs_status = hs_prepare(nullptr, re_pattern.c_str(), &database, &scratch);
+    if (hs_status.ok()) { // use hyperscan
         auto sz = val.size();
         for (size_t i = 0; i < sz; i++) {
             const auto& str_ref = val.get_data_at(i);
@@ -467,6 +470,9 @@ Status FunctionLikeBase::regexp_fn(const LikeSearchState* state, const ColumnStr
         hs_free_scratch(scratch);
         hs_free_database(database);
     } else { // fallback to re2
+        if (!state->enable_hyperscan_fallback) {
+            return hs_status;
+        }
         RE2::Options opts;
         opts.set_never_nl(false);
         opts.set_dot_nl(true);
@@ -487,8 +493,19 @@ Status FunctionLikeBase::regexp_fn(const LikeSearchState* state, const ColumnStr
 }
 
 // hyperscan compile expression to database and allocate scratch space
+bool FunctionLikeBase::should_fallback_to_re2(std::string_view regexp) {
+    return is_hyperscan_regexp_expensive(regexp);
+}
+
 Status FunctionLikeBase::hs_prepare(FunctionContext* context, const char* expression,
                                     hs_database_t** database, hs_scratch_t** scratch) {
+    if (should_fallback_to_re2(expression)) {
+        *database = nullptr;
+        *scratch = nullptr;
+        // Callers either fall back to RE2 or return this status based on the session variable.
+        return Status::RuntimeError<false>(HYPERSCAN_BOUNDED_REPEAT_ERROR);
+    }
+
     hs_compile_error_t* compile_err;
     auto res = hs_compile(expression, HS_FLAG_DOTALL | HS_FLAG_ALLOWEMPTY | HS_FLAG_UTF8,
                           HS_MODE_BLOCK, nullptr, database, &compile_err);
@@ -497,7 +514,7 @@ Status FunctionLikeBase::hs_prepare(FunctionContext* context, const char* expres
         *database = nullptr;
         std::string error_message = compile_err->message;
         hs_free_compile_error(compile_err);
-        // Do not call FunctionContext::set_error here, since we do not want to cancel the query here.
+        // Callers either fall back to RE2 or return this status based on the session variable.
         return Status::RuntimeError<false>("hs_compile regex pattern error:" + error_message);
     }
     hs_free_compile_error(compile_err);
@@ -506,7 +523,7 @@ Status FunctionLikeBase::hs_prepare(FunctionContext* context, const char* expres
         hs_free_database(*database);
         *database = nullptr;
         *scratch = nullptr;
-        // Do not call FunctionContext::set_error here, since we do not want to cancel the query here.
+        // Callers either fall back to RE2 or return this status based on the session variable.
         return Status::RuntimeError<false>("hs_alloc_scratch allocate scratch space error");
     }
 
@@ -942,12 +959,19 @@ Status FunctionLike::construct_like_const_state(FunctionContext* context, const
 
         hs_database_t* database = nullptr;
         hs_scratch_t* scratch = nullptr;
-        if (try_hyperscan && hs_prepare(context, re_pattern.c_str(), &database, &scratch).ok()) {
+        Status hs_status;
+        if (try_hyperscan) {
+            hs_status = hs_prepare(context, re_pattern.c_str(), &database, &scratch);
+        }
+        if (try_hyperscan && hs_status.ok()) {
             // use hyperscan
             state->search_state.hs_database.reset(database);
             state->search_state.hs_scratch.reset(scratch);
         } else {
             // fallback to re2
+            if (try_hyperscan && !state->search_state.enable_hyperscan_fallback) {
+                return hs_status;
+            }
             // reset hs_database to nullptr to indicate not use hyperscan
             state->search_state.hs_database.reset();
             state->search_state.hs_scratch.reset();
@@ -974,6 +998,8 @@ Status FunctionLike::open(FunctionContext* context, FunctionContext::FunctionSta
     }
     std::shared_ptr<LikeState> state = std::make_shared<LikeState>();
     state->is_like_pattern = true;
+    state->search_state.enable_hyperscan_fallback =
+            context->state()->query_options().enable_hyperscan_fallback;
     state->function = like_fn;
     state->scalar_function = like_fn_scalar;
     if (context->is_col_constant(2)) {
@@ -1004,6 +1030,8 @@ Status FunctionRegexpLike::open(FunctionContext* context,
     std::shared_ptr<LikeState> state = std::make_shared<LikeState>();
     context->set_function_state(scope, state);
     state->is_like_pattern = false;
+    state->search_state.enable_hyperscan_fallback =
+            context->state()->query_options().enable_hyperscan_fallback;
     state->function = regexp_fn;
     state->scalar_function = regexp_fn_scalar;
     if (context->is_col_constant(1)) {
@@ -1035,12 +1063,16 @@ Status FunctionRegexpLike::open(FunctionContext* context,
         } else {
             hs_database_t* database = nullptr;
             hs_scratch_t* scratch = nullptr;
-            if (hs_prepare(context, pattern_str.c_str(), &database, &scratch).ok()) {
+            auto hs_status = hs_prepare(context, pattern_str.c_str(), &database, &scratch);
+            if (hs_status.ok()) {
                 // use hyperscan
                 state->search_state.hs_database.reset(database);
                 state->search_state.hs_scratch.reset(scratch);
             } else {
                 // fallback to re2
+                if (!state->search_state.enable_hyperscan_fallback) {
+                    return hs_status;
+                }
                 // reset hs_database to nullptr to indicate not use hyperscan
                 state->search_state.hs_database.reset();
                 state->search_state.hs_scratch.reset();
diff --git a/be/src/exprs/function/like.h b/be/src/exprs/function/like.h
index 461c97956bcc7f..d648919363dce2 100644
--- a/be/src/exprs/function/like.h
+++ b/be/src/exprs/function/like.h
@@ -29,6 +29,7 @@
 #include <functional>
 #include <memory>
 #include <string>
+#include <string_view>
 
 #include "common/status.h"
 #include "core/block/column_numbers.h"
@@ -182,6 +183,8 @@ struct LikeSearchState {
 
     std::string pattern_str;
 
+    bool enable_hyperscan_fallback = true;
+
     /// Used for LIKE predicates if the pattern is a constant argument, and is either a
     /// constant string or has a constant string at the beginning or end of the pattern.
     /// This will be set in order to check for that pattern in the corresponding part of
@@ -226,7 +229,7 @@ struct LikeSearchState {
 
     LikeSearchState() = default;
 
-    Status clone(LikeSearchState& cloned);
+    Status clone(LikeSearchState& cloned) const;
 
     void set_search_string(const std::string& search_string_arg) {
         search_string = search_string_arg;
@@ -292,6 +295,8 @@ class FunctionLikeBase : public IFunction {
     friend struct VectorEndsWithSearchState;
 
 protected:
+    static bool should_fallback_to_re2(std::string_view regexp);
+
     Status vector_const(const ColumnString& values, const StringRef* pattern_val,
                         ColumnUInt8::Container& result, const LikeFn& function,
                         LikeSearchState* search_state) const;
diff --git a/be/src/exprs/function/match.cpp b/be/src/exprs/function/match.cpp
index a280dd035e25b4..e9048ab7afa1d1 100644
--- a/be/src/exprs/function/match.cpp
+++ b/be/src/exprs/function/match.cpp
@@ -25,6 +25,7 @@
 #include "storage/index/index_reader_helper.h"
 #include "storage/index/inverted/analyzer/analyzer.h"
 #include "util/debug_points.h"
+#include "util/hyperscan_util.h"
 
 namespace doris {
 
@@ -507,6 +508,10 @@ Status FunctionMatchRegexp::execute_match(FunctionContext* context, const std::s
     hs_compile_error_t* compile_err = nullptr;
     hs_scratch_t* scratch = nullptr;
 
+    if (is_hyperscan_regexp_expensive(pattern)) {
+        return Status::Error<ErrorCode::INDEX_INVALID_PARAMETERS>(HYPERSCAN_BOUNDED_REPEAT_ERROR);
+    }
+
     if (hs_compile(pattern.data(), HS_FLAG_DOTALL | HS_FLAG_ALLOWEMPTY | HS_FLAG_UTF8,
                    HS_MODE_BLOCK, nullptr, &database, &compile_err) != HS_SUCCESS) {
         std::string err_message = "hyperscan compilation failed: ";
diff --git a/be/src/exprs/function/regexps.h b/be/src/exprs/function/regexps.h
index d521c7d9decd73..15520d448c2336 100644
--- a/be/src/exprs/function/regexps.h
+++ b/be/src/exprs/function/regexps.h
@@ -33,6 +33,7 @@
 
 #include "common/exception.h"
 #include "core/string_ref.h"
+#include "util/hyperscan_util.h"
 
 namespace doris::multiregexps {
 
@@ -144,6 +145,9 @@ Regexps constructRegexps(const std::vector<String>& str_patterns,
 
     for (auto& pattern : patterns) {
         LOG(INFO) << "pattern: " << pattern << "\n";
+        if (is_hyperscan_regexp_expensive(pattern)) {
+            throw doris::Exception(Status::InvalidArgument(HYPERSCAN_BOUNDED_REPEAT_ERROR));
+        }
     }
 
     hs_error_t err;
diff --git a/be/src/storage/index/inverted/query/regexp_query.cpp b/be/src/storage/index/inverted/query/regexp_query.cpp
index 7002f21446895d..fe9e3666ea547e 100644
--- a/be/src/storage/index/inverted/query/regexp_query.cpp
+++ b/be/src/storage/index/inverted/query/regexp_query.cpp
@@ -23,6 +23,7 @@
 
 #include "common/logging.h"
 #include "util/debug_points.h"
+#include "util/hyperscan_util.h"
 
 namespace doris::segment_v2 {
 
@@ -39,6 +40,9 @@ void RegexpQuery::add(const InvertedIndexQueryInfo& query_info) {
     }
 
     const std::string& pattern = query_info.term_infos[0].get_single_term();
+    if (is_hyperscan_regexp_expensive(pattern)) {
+        throw Exception(ErrorCode::INVALID_ARGUMENT, HYPERSCAN_BOUNDED_REPEAT_ERROR);
+    }
     auto prefix = get_regex_prefix(pattern);
 
     hs_database_t* database = nullptr;
diff --git a/be/src/storage/index/inverted/query_v2/regexp_query/regexp_weight.cpp b/be/src/storage/index/inverted/query_v2/regexp_query/regexp_weight.cpp
index ac6a905ba473cc..5ecfad0d3ae14a 100644
--- a/be/src/storage/index/inverted/query_v2/regexp_query/regexp_weight.cpp
+++ b/be/src/storage/index/inverted/query_v2/regexp_query/regexp_weight.cpp
@@ -41,6 +41,7 @@
 #include "storage/index/inverted/query_v2/nullable_scorer.h"
 #include "storage/index/inverted/query_v2/segment_postings.h"
 #include "storage/index/inverted/util/string_helper.h"
+#include "util/hyperscan_util.h"
 
 CL_NS_USE(index)
 
@@ -70,6 +71,9 @@ ScorerPtr RegexpWeight::scorer(const QueryExecutionContext& context,
 
 ScorerPtr RegexpWeight::regexp_scorer(const QueryExecutionContext& context,
                                       const std::string& binding_key) {
+    if (is_hyperscan_regexp_expensive(_pattern)) {
+        throw Exception(ErrorCode::INVALID_ARGUMENT, HYPERSCAN_BOUNDED_REPEAT_ERROR);
+    }
     auto prefix = get_regex_prefix(_pattern);
 
     hs_database_t* database = nullptr;
diff --git a/be/src/util/hyperscan_util.cpp b/be/src/util/hyperscan_util.cpp
new file mode 100644
index 00000000000000..570bbf83d47e08
--- /dev/null
+++ b/be/src/util/hyperscan_util.cpp
@@ -0,0 +1,139 @@
+// 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 "util/hyperscan_util.h"
+
+#include <re2/re2.h>
+#include <re2/stringpiece.h>
+
+#include <charconv>
+#include <string>
+
+namespace doris {
+namespace {
+
+bool is_larger_than_fifty(std::string_view str) {
+    int number = 0;
+    auto [_, error] = std::from_chars(str.data(), str.data() + str.size(), number);
+    return error == std::errc() && number > 50;
+}
+
+std::string mask_escaped_characters_and_character_classes(std::string_view regexp) {
+    std::string masked_regexp(regexp);
+    bool escaped = false;
+    bool in_character_class = false;
+    bool character_class_can_close = false;
+    bool character_class_can_negate = false;
+    for (char& masked_character : masked_regexp) {
+        const char current = masked_character;
+        if (escaped) {
+            masked_character = ' ';
+            escaped = false;
+            if (in_character_class) {
+                character_class_can_close = true;
+                character_class_can_negate = false;
+            }
+            continue;
+        }
+        if (current == '\\') {
+            masked_character = ' ';
+            escaped = true;
+            continue;
+        }
+        if (in_character_class) {
+            masked_character = ' ';
+            if (current == ']' && character_class_can_close) {
+                in_character_class = false;
+            } else if (current == '^' && character_class_can_negate) {
+                character_class_can_negate = false;
+            } else {
+                character_class_can_close = true;
+                character_class_can_negate = false;
+            }
+            continue;
+        }
+        if (current == '[') {
+            masked_character = ' ';
+            in_character_class = true;
+            character_class_can_close = false;
+            character_class_can_negate = true;
+        }
+    }
+    return masked_regexp;
+}
+
+class SlowWithHyperscanChecker {
+public:
+    SlowWithHyperscanChecker()
+            : _searcher_one_repeat(R"(\{\s*([\d]+)\s*,?\s*})"),
+              _searcher_two_repeats(R"(\{\s*([\d]+)\s*,\s*([\d]+)\s*\})") {}
+
+    bool is_slow(std::string_view regexp) const {
+        const std::string masked_regexp = mask_escaped_characters_and_character_classes(regexp);
+        return is_slow_one_repeat(masked_regexp) || is_slow_two_repeats(masked_regexp);
+    }
+
+private:
+    bool is_slow_one_repeat(std::string_view regexp) const {
+        re2::StringPiece haystack(regexp.data(), regexp.size());
+        re2::StringPiece matches[2];
+        size_t start_pos = 0;
+        while (start_pos < haystack.size()) {
+            if (!_searcher_one_repeat.Match(haystack, start_pos, haystack.size(),
+                                            re2::RE2::Anchor::UNANCHORED, matches, 2)) {
+                break;
+            }
+
+            start_pos = matches[0].data() - haystack.data() + matches[0].size();
+            if (is_larger_than_fifty({matches[1].data(), matches[1].size()})) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    bool is_slow_two_repeats(std::string_view regexp) const {
+        re2::StringPiece haystack(regexp.data(), regexp.size());
+        re2::StringPiece matches[3];
+        size_t start_pos = 0;
+        while (start_pos < haystack.size()) {
+            if (!_searcher_two_repeats.Match(haystack, start_pos, haystack.size(),
+                                             re2::RE2::Anchor::UNANCHORED, matches, 3)) {
+                break;
+            }
+
+            start_pos = matches[0].data() - haystack.data() + matches[0].size();
+            if (is_larger_than_fifty({matches[1].data(), matches[1].size()}) ||
+                is_larger_than_fifty({matches[2].data(), matches[2].size()})) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    re2::RE2 _searcher_one_repeat;
+    re2::RE2 _searcher_two_repeats;
+};
+
+} // namespace
+
+bool is_hyperscan_regexp_expensive(std::string_view regexp) {
+    static const SlowWithHyperscanChecker slow_with_hyperscan_checker;
+    return slow_with_hyperscan_checker.is_slow(regexp);
+}
+
+} // namespace doris
diff --git a/be/src/util/hyperscan_util.h b/be/src/util/hyperscan_util.h
new file mode 100644
index 00000000000000..27ef9fdfe177cd
--- /dev/null
+++ b/be/src/util/hyperscan_util.h
@@ -0,0 +1,29 @@
+// 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.
+
+#pragma once
+
+#include <string_view>
+
+namespace doris {
+
+inline constexpr std::string_view HYPERSCAN_BOUNDED_REPEAT_ERROR =
+        "Skip hyperscan compilation because bounded repetition exceeds 50";
+
+bool is_hyperscan_regexp_expensive(std::string_view regexp);
+
+} // namespace doris
diff --git a/be/test/exprs/function/function_like_test.cpp b/be/test/exprs/function/function_like_test.cpp
index 82618a790e99e7..45f2b3af91dde2 100644
--- a/be/test/exprs/function/function_like_test.cpp
+++ b/be/test/exprs/function/function_like_test.cpp
@@ -16,8 +16,12 @@
 // under the License.
 
 #include <cstdint>
+#include <memory>
 #include <string>
+#include <vector>
 
+#include "core/block/block.h"
+#include "core/column/column_const.h"
 #include "core/column/column_string.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/data_type_array.h"
@@ -27,11 +31,51 @@
 #include "core/types.h"
 #include "exprs/function/function_test_util.h"
 #include "exprs/function/like.h"
+#include "exprs/function_context.h"
 #include "gtest/gtest_pred_impl.h"
+#include "runtime/runtime_state.h"
 #include "testutil/any_type.h"
 
 namespace doris {
 
+class FunctionLikeTestHelper : public FunctionLikeBase {
+public:
+    using FunctionLikeBase::should_fallback_to_re2;
+};
+
+template <typename Function>
+Status execute_pattern_with_fallback_disabled(const std::string& value, const std::string& pattern,
+                                              bool constant_known_at_open) {
+    TQueryOptions query_options;
+    query_options.__set_enable_hyperscan_fallback(false);
+    RuntimeState runtime_state(query_options, TQueryGlobals {});
+
+    auto string_type = std::make_shared<DataTypeString>();
+    auto context = FunctionContext::create_context(
+            &runtime_state, std::make_shared<DataTypeUInt8>(), {string_type, string_type});
+
+    auto values = ColumnString::create();
+    values->insert_data(value.data(), value.size());
+    auto patterns = ColumnString::create();
+    patterns->insert_data(pattern.data(), pattern.size());
+    ColumnPtr pattern_column = ColumnConst::create(std::move(patterns), 1);
+
+    std::vector<std::shared_ptr<ColumnPtrWrapper>> constant_columns(2);
+    if (constant_known_at_open) {
+        constant_columns[1] = std::make_shared<ColumnPtrWrapper>(pattern_column);
+    }
+    context->set_constant_cols(constant_columns);
+
+    Function function;
+    RETURN_IF_ERROR(function.open(context.get(), FunctionContext::THREAD_LOCAL));
+
+    Block block;
+    block.insert({std::move(values), string_type, "value"});
+    block.insert({std::move(pattern_column), string_type, "pattern"});
+    block.insert({nullptr, std::make_shared<DataTypeUInt8>(), "result"});
+    return function.execute_impl(context.get(), block, {0, 1}, 2, 1);
+}
+
 TEST(FunctionLikeTest, like) {
     std::string func_name = "like";
 
@@ -137,6 +181,65 @@ TEST(FunctionLikeTest, regexp) {
     }
 }
 
+TEST(FunctionLikeTest, hyperscan_bounded_repeat_fallback) {
+    std::string func_name = "regexp";
+    std::string matching_value = "prompt_rewrite.h03" + std::string(500, 'x') + "429";
+
+    DataSet data_set = {
+            {{matching_value, std::string(R"(prompt_rewrite\.h03.{0,1000}429)")}, uint8_t(1)},
+            {{std::string("prompt_rewrite.h03") + std::string(500, 'x') + "430",
+              std::string(R"(prompt_rewrite\.h03.{0,1000}429)")},
+             uint8_t(0)}};
+
+    InputTypeSet input_types = {PrimitiveType::TYPE_VARCHAR, PrimitiveType::TYPE_VARCHAR};
+    check_function_all_arg_comb<DataTypeUInt8, true>(func_name, input_types, data_set);
+}
+
+TEST(FunctionLikeTest, hyperscan_bounded_repeat_threshold) {
+    EXPECT_FALSE(FunctionLikeTestHelper::should_fallback_to_re2("a*"));
+    EXPECT_FALSE(FunctionLikeTestHelper::should_fallback_to_re2("a{50}"));
+    EXPECT_FALSE(FunctionLikeTestHelper::should_fallback_to_re2("a{0,50}"));
+    EXPECT_TRUE(FunctionLikeTestHelper::should_fallback_to_re2("a{51}"));
+    EXPECT_TRUE(FunctionLikeTestHelper::should_fallback_to_re2("a{51,}"));
+    EXPECT_TRUE(FunctionLikeTestHelper::should_fallback_to_re2("a{0,51}"));
+    EXPECT_TRUE(FunctionLikeTestHelper::should_fallback_to_re2("a{51,51}"));
+    EXPECT_TRUE(FunctionLikeTestHelper::should_fallback_to_re2("a{ 0, 1000 }"));
+    EXPECT_FALSE(FunctionLikeTestHelper::should_fallback_to_re2(R"(a\{51\})"));
+    EXPECT_FALSE(FunctionLikeTestHelper::should_fallback_to_re2("[a{51}]"));
+    EXPECT_TRUE(FunctionLikeTestHelper::should_fallback_to_re2("[^^](ab?c?d){1000,5000}"));
+}
+
+TEST(FunctionLikeTest, hyperscan_bounded_repeat_fallback_disabled) {
+    for (bool constant_known_at_open : {false, true}) {
+        SCOPED_TRACE(constant_known_at_open ? "prepare during open" : "prepare during execute");
+        auto status = execute_pattern_with_fallback_disabled<FunctionRegexpLike>(
+                "prompt_rewrite.h03xxx429", R"(prompt_rewrite\.h03.{0,1000}429)",
+                constant_known_at_open);
+        EXPECT_FALSE(status.ok());
+        EXPECT_NE(status.to_string().find("bounded repetition exceeds 50"), std::string::npos);
+
+        status = execute_pattern_with_fallback_disabled<FunctionRegexpLike>(
+                "^abc", "[^^](ab?c?d){1000,5000}", constant_known_at_open);
+        EXPECT_FALSE(status.ok());
+        EXPECT_NE(status.to_string().find("bounded repetition exceeds 50"), std::string::npos);
+    }
+}
+
+TEST(FunctionLikeTest, hyperscan_bounded_repeat_literal_with_fallback_disabled) {
+    for (bool constant_known_at_open : {false, true}) {
+        SCOPED_TRACE(constant_known_at_open ? "prepare during open" : "prepare during execute");
+        EXPECT_TRUE(execute_pattern_with_fallback_disabled<FunctionRegexpLike>(
+                            "a{51}", R"(a\{51\})", constant_known_at_open)
+                            .ok());
+        EXPECT_TRUE(execute_pattern_with_fallback_disabled<FunctionRegexpLike>(
+                            "a", "[a{51}]", constant_known_at_open)
+                            .ok());
+        EXPECT_TRUE(execute_pattern_with_fallback_disabled<FunctionLike>("a{51}", "_{51}",
+                                                                         constant_known_at_open)
+                            .ok());
+    }
+}
+
 TEST(FunctionLikeTest, regexp_extract) {
     std::string func_name = "regexp_extract";
 
diff --git a/be/test/exprs/function/function_match_test.cpp b/be/test/exprs/function/function_match_test.cpp
index 738381a04ef576..fccc2ee80df53b 100644
--- a/be/test/exprs/function/function_match_test.cpp
+++ b/be/test/exprs/function/function_match_test.cpp
@@ -28,6 +28,7 @@
 #include "core/column/column_string.h"
 #include "core/column/column_vector.h"
 #include "exprs/function/match.h"
+#include "runtime/runtime_state.h"
 #include "storage/index/inverted/analyzer/analyzer.h"
 
 namespace doris {
@@ -85,6 +86,23 @@ TEST(FunctionMatchTest, analyse_query_str) {
     }
 }
 
+TEST(FunctionMatchTest, regexp_rejects_expensive_bounded_repeat) {
+    TQueryOptions query_options;
+    query_options.__set_enable_match_without_inverted_index(true);
+    RuntimeState runtime_state(query_options, TQueryGlobals {});
+    auto context = FunctionContext::create_context(&runtime_state, {}, {});
+
+    auto string_col = ColumnString::create();
+    string_col->insert_data("abcd", 4);
+    ColumnUInt8::Container result(1, 0);
+
+    FunctionMatchRegexp function;
+    Status status = function.execute_match(context.get(), "test_column", "(ab?c?d){1000,5000}", 1,
+                                           string_col.get(), nullptr, nullptr, result);
+    EXPECT_FALSE(status.ok());
+    EXPECT_NE(status.to_string().find("bounded repetition exceeds 50"), std::string::npos);
+}
+
 // Test FunctionMatchAny::execute_match
 TEST(FunctionMatchTest, match_any_execute) {
     FunctionMatchAny func_match_any;
@@ -841,4 +859,4 @@ TEST(FunctionMatchTest, function_registration) {
     EXPECT_TRUE(true);
 }
 
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/test/exprs/function/function_multi_match_test.cpp b/be/test/exprs/function/function_multi_match_test.cpp
index d184dd5afa5b41..8008e877acc476 100644
--- a/be/test/exprs/function/function_multi_match_test.cpp
+++ b/be/test/exprs/function/function_multi_match_test.cpp
@@ -22,6 +22,7 @@
 #include "core/block/column_with_type_and_name.h"
 #include "core/block/columns_with_type_and_name.h"
 #include "core/data_type/data_type_string.h"
+#include "exprs/function/regexps.h"
 #include "storage/index/inverted/inverted_index_reader.h"
 
 namespace doris {
@@ -73,4 +74,9 @@ TEST_F(FunctionMultiMatchTest, EvaluateInvertedIndexWithNullIterator) {
             << "Error message should contain column name. Actual message: " << error_msg;
 }
 
+TEST_F(FunctionMultiMatchTest, RejectsExpensiveBoundedRepeat) {
+    std::vector<String> patterns = {"(ab?c?d){1000,5000}"};
+    EXPECT_THROW((multiregexps::constructRegexps<false, false>(patterns, std::nullopt)), Exception);
+}
+
 } // namespace doris
diff --git a/be/test/storage/index/inverted/query/regexp_query_test.cpp b/be/test/storage/index/inverted/query/regexp_query_test.cpp
index e212592d2f317b..4d07f3479202a2 100644
--- a/be/test/storage/index/inverted/query/regexp_query_test.cpp
+++ b/be/test/storage/index/inverted/query/regexp_query_test.cpp
@@ -218,6 +218,27 @@ TEST_F(RegexpQueryTest, AddWithInvalidTermsSize) {
     }
 }
 
+TEST_F(RegexpQueryTest, AddRejectsExpensiveBoundedRepeat) {
+    std::shared_ptr<lucene::search::IndexSearcher> searcher = nullptr;
+    OlapReaderStatistics stats;
+    RuntimeState runtime_state;
+    TQueryOptions query_options;
+    query_options.inverted_index_max_expansions = 50;
+    runtime_state.set_query_options(query_options);
+    io::IOContext io_ctx;
+
+    auto context = std::make_shared<IndexQueryContext>();
+    context->io_ctx = &io_ctx;
+    context->runtime_state = &runtime_state;
+    context->stats = &stats;
+    RegexpQuery regexp_query(searcher, context);
+
+    InvertedIndexQueryInfo query_info;
+    query_info.field_name = L"test_field";
+    query_info.term_infos.push_back({"(ab?c?d){1000,5000}", 0});
+    EXPECT_THROW(regexp_query.add(query_info), Exception);
+}
+
 TEST_F(RegexpQueryTest, AddWithInvalidPattern) {
     // Create a mock searcher and query options for testing
     std::shared_ptr<lucene::search::IndexSearcher> searcher = nullptr;
@@ -422,4 +443,4 @@ TEST_F(RegexpQueryTest, AddWithBackreferencePattern) {
     EXPECT_NO_THROW(regexp_query.add(query_info));
 }
 
-} // namespace doris::segment_v2
\ No newline at end of file
+} // namespace doris::segment_v2
diff --git a/be/test/storage/index/inverted/query_v2/regexp_query_test.cpp b/be/test/storage/index/inverted/query_v2/regexp_query_test.cpp
index 397ecdae63b784..9e2567b298fc37 100644
--- a/be/test/storage/index/inverted/query_v2/regexp_query_test.cpp
+++ b/be/test/storage/index/inverted/query_v2/regexp_query_test.cpp
@@ -133,6 +133,19 @@ TEST_F(RegexpQueryV2Test, test_regexp_query_construction) {
     ASSERT_NE(regexp_weight, nullptr);
 }
 
+TEST_F(RegexpQueryV2Test, test_rejects_expensive_bounded_repeat) {
+    auto context = std::make_shared<IndexQueryContext>();
+    context->collection_statistics = std::make_shared<CollectionStatistics>();
+    context->collection_similarity = std::make_shared<CollectionSimilarity>();
+
+    std::wstring field = StringHelper::to_wstring("content");
+    auto query = std::make_shared<query_v2::RegexpQuery>(context, field, "(ab?c?d){1000,5000}");
+    auto weight = query->weight(false);
+    query_v2::QueryExecutionContext exec_ctx;
+
+    EXPECT_THROW(weight->scorer(exec_ctx), Exception);
+}
+
 // Test regexp query with scoring enabled
 TEST_F(RegexpQueryV2Test, test_regexp_query_with_scoring) {
     auto context = std::make_shared<IndexQueryContext>();
@@ -565,4 +578,4 @@ TEST_F(RegexpQueryV2Test, test_make_exact_match_wildcard_pattern) {
     _CLDECDELETE(dir);
 }
 
-} // namespace doris::segment_v2
\ No newline at end of file
+} // namespace doris::segment_v2
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/EnvFactory.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/EnvFactory.java
index e933f8c9130d82..993389d836f0d2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/EnvFactory.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/EnvFactory.java
@@ -174,7 +174,8 @@ protected static boolean hasNereidsDistributedPlans(NereidsPlanner planner) {
     // Used for broker load task/export task/update coordinator
     public Coordinator createCoordinator(Long jobId, TUniqueId queryId, DescriptorTable descTable,
                                          List<PlanFragment> fragments, List<ScanNode> scanNodes,
-                                         String timezone, boolean loadZeroTolerance, boolean enableProfile) {
+                                         String timezone, boolean loadZeroTolerance, boolean enableProfile,
+                                         boolean enableHyperscanFallback) {
         if (SessionVariable.canUseNereidsDistributePlanner()) {
             if (queryId == null) {
                 UUID taskId = UUID.randomUUID();
@@ -200,11 +201,12 @@ public Coordinator createCoordinator(Long jobId, TUniqueId queryId, DescriptorTa
 
             return new NereidsCoordinator(
                     jobId, queryId, descTable, fragments, distributedPlans.valueList(),
-                    scanNodes, timezone, loadZeroTolerance, enableProfile
+                    scanNodes, timezone, loadZeroTolerance, enableProfile, enableHyperscanFallback
             );
         }
         return new Coordinator(
-                jobId, queryId, descTable, fragments, scanNodes, timezone, loadZeroTolerance, enableProfile
+                jobId, queryId, descTable, fragments, scanNodes, timezone, loadZeroTolerance, enableProfile,
+                enableHyperscanFallback
         );
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnvFactory.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnvFactory.java
index b3d469c41ff160..b789d284cb88f1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnvFactory.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudEnvFactory.java
@@ -176,13 +176,15 @@ public Coordinator createCoordinator(ConnectContext context, Planner planner,
     @Override
     public Coordinator createCoordinator(Long jobId, TUniqueId queryId, DescriptorTable descTable,
                                          List<PlanFragment> fragments, List<ScanNode> scanNodes,
-                                         String timezone, boolean loadZeroTolerance, boolean enableProfile) {
+                                         String timezone, boolean loadZeroTolerance, boolean enableProfile,
+                                         boolean enableHyperscanFallback) {
         if (SessionVariable.canUseNereidsDistributePlanner()) {
             return super.createCoordinator(
-                    jobId, queryId, descTable, fragments, scanNodes, timezone, loadZeroTolerance, enableProfile);
+                    jobId, queryId, descTable, fragments, scanNodes, timezone, loadZeroTolerance, enableProfile,
+                    enableHyperscanFallback);
         }
         return new CloudCoordinator(jobId, queryId, descTable, fragments, scanNodes, timezone, loadZeroTolerance,
-                                enableProfile);
+                                enableProfile, enableHyperscanFallback);
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java
index cd5c8dd1b0c52f..c5d77d3f7978e5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java
@@ -45,6 +45,7 @@
 import org.apache.doris.qe.Coordinator;
 import org.apache.doris.qe.OriginStatement;
 import org.apache.doris.qe.QeProcessorImpl;
+import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.qe.StmtExecutor;
 import org.apache.doris.resource.computegroup.ComputeGroupMgr;
 import org.apache.doris.system.Backend;
@@ -161,7 +162,9 @@ brokerFileGroups, getDeadlineMs(), getExecMemLimit(),
                 transactionId, this, getTimeZone(), getTimeout(),
                 getLoadParallelism(), getSendBatchParallelism(),
                 getMaxFilterRatio() <= 0, enableProfile ? jobProfile : null, isSingleTabletLoadPerSink(),
-                getPriority(), isEnableMemtableOnSinkNode, batchSize, cloudClusterId);
+                getPriority(), isEnableMemtableOnSinkNode, batchSize, cloudClusterId,
+                Boolean.parseBoolean(sessionVariables.getOrDefault(
+                        SessionVariable.ENABLE_HYPERSCAN_FALLBACK, Boolean.toString(true))));
         UUID uuid = UUID.randomUUID();
         TUniqueId loadId = new TUniqueId(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits());
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudLoadLoadingTask.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudLoadLoadingTask.java
index 607a68b2e70c81..12351f734e878b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudLoadLoadingTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudLoadLoadingTask.java
@@ -51,10 +51,11 @@ public CloudLoadLoadingTask(UserIdentity userinfo, Database db, OlapTable table,
             long timeoutS, int loadParallelism, int sendBatchParallelism,
             boolean loadZeroTolerance, Profile jobProfile, boolean singleTabletLoadPerSink,
             Priority priority, boolean enableMemTableOnSinkNode, int batchSize,
-            String clusterId) {
+            String clusterId, boolean enableHyperscanFallback) {
         super(userinfo, db, table, brokerDesc, fileGroups, jobDeadlineMs, execMemLimit, strictMode, isPartialUpdate,
                 partialUpdateNewKeyPolicy, txnId, callback, timezone, timeoutS, loadParallelism, sendBatchParallelism,
-                loadZeroTolerance, jobProfile, singleTabletLoadPerSink, priority, enableMemTableOnSinkNode, batchSize);
+                loadZeroTolerance, jobProfile, singleTabletLoadPerSink, priority, enableMemTableOnSinkNode, batchSize,
+                enableHyperscanFallback);
         this.cloudClusterId = clusterId;
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/qe/CloudCoordinator.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/qe/CloudCoordinator.java
index 39eb6d36ede000..bca0e9598fe462 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/qe/CloudCoordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/qe/CloudCoordinator.java
@@ -52,8 +52,9 @@ public CloudCoordinator(ConnectContext context,
 
     public CloudCoordinator(Long jobId, TUniqueId queryId, DescriptorTable descTable, List<PlanFragment> fragments,
                        List<ScanNode> scanNodes, String timezone, boolean loadZeroTolerance,
-                    boolean enbaleProfile) {
-        super(jobId, queryId, descTable, fragments, scanNodes, timezone, loadZeroTolerance, enbaleProfile);
+                    boolean enbaleProfile, boolean enableHyperscanFallback) {
+        super(jobId, queryId, descTable, fragments, scanNodes, timezone, loadZeroTolerance, enbaleProfile,
+                enableHyperscanFallback);
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
index 1be1a5003973d3..f8a042d804668f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java
@@ -269,7 +269,8 @@ public void generatePlan(OlapTable table) throws UserException {
                     "get table read lock timeout, database=" + request.getDb() + ",table=" + table.getName());
         }
         try {
-            NereidsStreamLoadTask streamLoadTask = NereidsStreamLoadTask.fromTStreamLoadPutRequest(request);
+            NereidsStreamLoadTask streamLoadTask = NereidsStreamLoadTask.fromTStreamLoadPutRequest(
+                    request, ConnectContext.get().getSessionVariable().enableHyperscanFallback);
             if (isMultiTableRequest) {
                 buildMultiTableStreamLoadTask(streamLoadTask, request.getTxnId());
             }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
index c7b9415c68d6bb..af493f6f1ea7aa 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
@@ -308,7 +308,9 @@ brokerFileGroups, getDeadlineMs(), getExecMemLimit(),
                 transactionId, this, getTimeZone(), getTimeout(),
                 getLoadParallelism(), getSendBatchParallelism(),
                 getMaxFilterRatio() <= 0, enableProfile ? jobProfile : null, isSingleTabletLoadPerSink(),
-                getPriority(), isEnableMemtableOnSinkNode, batchSize);
+                getPriority(), isEnableMemtableOnSinkNode, batchSize,
+                Boolean.parseBoolean(sessionVariables.getOrDefault(
+                        SessionVariable.ENABLE_HYPERSCAN_FALLBACK, Boolean.toString(true))));
 
         UUID uuid = UUID.randomUUID();
         TUniqueId loadId = new TUniqueId(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BulkLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BulkLoadJob.java
index fbc34895a43ca7..c054978bde627f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BulkLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BulkLoadJob.java
@@ -108,10 +108,13 @@ public BulkLoadJob(EtlJobType jobType, long dbId, String label,
             sessionVariables.put(SessionVariable.AUTO_PROFILE_THRESHOLD_MS,
                                     Long.toString(var.getAutoProfileThresholdMs()));
             sessionVariables.put(SessionVariable.PROFILE_LEVEL, Long.toString(var.getProfileLevel()));
+            sessionVariables.put(SessionVariable.ENABLE_HYPERSCAN_FALLBACK,
+                    Boolean.toString(var.enableHyperscanFallback));
         } else {
             sessionVariables.put(SessionVariable.SQL_MODE, String.valueOf(SqlModeHelper.MODE_DEFAULT));
             sessionVariables.put(SessionVariable.AUTO_PROFILE_THRESHOLD_MS, Long.toString(-1));
             sessionVariables.put(SessionVariable.PROFILE_LEVEL, Long.toString(1));
+            sessionVariables.put(SessionVariable.ENABLE_HYPERSCAN_FALLBACK, Boolean.toString(true));
         }
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
index 2e9acff237b67d..e1894f88abcd56 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
@@ -89,6 +89,7 @@ public class LoadLoadingTask extends LoadTask {
     private long beginTime;
 
     private List<TPipelineWorkloadGroup> tWorkloadGroups = null;
+    private final boolean enableHyperscanFallback;
 
     protected UserIdentity userInfo;
 
@@ -99,7 +100,8 @@ public LoadLoadingTask(UserIdentity userInfo, Database db, OlapTable table,
             long txnId, LoadTaskCallback callback, String timezone,
             long timeoutS, int loadParallelism, int sendBatchParallelism,
             boolean loadZeroTolerance, Profile jobProfile, boolean singleTabletLoadPerSink,
-            Priority priority, boolean enableMemTableOnSinkNode, int batchSize) {
+            Priority priority, boolean enableMemTableOnSinkNode, int batchSize,
+            boolean enableHyperscanFallback) {
         super(callback, TaskType.LOADING, priority);
         this.userInfo = userInfo;
         this.db = db;
@@ -123,6 +125,7 @@ public LoadLoadingTask(UserIdentity userInfo, Database db, OlapTable table,
         this.singleTabletLoadPerSink = singleTabletLoadPerSink;
         this.enableMemTableOnSinkNode = enableMemTableOnSinkNode;
         this.batchSize = batchSize;
+        this.enableHyperscanFallback = enableHyperscanFallback;
     }
 
     public void init(TUniqueId loadId, List<List<TBrokerFileStatus>> fileStatusList,
@@ -134,7 +137,8 @@ public void init(TUniqueId loadId, List<List<TBrokerFileStatus>> fileStatusList,
         }
         planner = new NereidsLoadingTaskPlanner(callback.getCallbackId(), txnId, db.getId(), table, brokerDesc,
                 brokerFileGroups, strictMode, isPartialUpdate, partialUpdateNewKeyPolicy, timezone, timeoutS,
-                loadParallelism, sendBatchParallelism, userInfo, singleTabletLoadPerSink, enableMemTableOnSinkNode);
+                loadParallelism, sendBatchParallelism, userInfo, singleTabletLoadPerSink, enableMemTableOnSinkNode,
+                enableHyperscanFallback);
         planner.plan(loadId, fileStatusList, fileNum);
     }
 
@@ -162,7 +166,7 @@ protected void executeOnce() throws Exception {
         Coordinator curCoordinator =  EnvFactory.getInstance().createCoordinator(callback.getCallbackId(),
                 loadId, planner.getDescTable(),
                 planner.getFragments(), planner.getScanNodes(), planner.getTimezone(), loadZeroTolerance,
-                enableProfile);
+                enableProfile, enableHyperscanFallback);
         if (enableProfile) {
             this.jobProfile.addExecutionProfile(curCoordinator.getExecutionProfile());
         }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
index 9873368f405114..14314c9b961caf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
@@ -320,6 +320,8 @@ public RoutineLoadJob(Long id, String name,
         if (ConnectContext.get() != null) {
             SessionVariable var = ConnectContext.get().getSessionVariable();
             sessionVariables.put(SessionVariable.SQL_MODE, Long.toString(var.getSqlMode()));
+            sessionVariables.put(SessionVariable.ENABLE_HYPERSCAN_FALLBACK,
+                    Boolean.toString(var.enableHyperscanFallback));
             this.memtableOnSinkNode = ConnectContext.get().getSessionVariable().enableMemtableOnSinkNode;
             if (Config.isCloudMode()) {
                 try {
@@ -330,9 +332,16 @@ public RoutineLoadJob(Long id, String name,
             }
         } else {
             sessionVariables.put(SessionVariable.SQL_MODE, String.valueOf(SqlModeHelper.MODE_DEFAULT));
+            sessionVariables.put(SessionVariable.ENABLE_HYPERSCAN_FALLBACK, Boolean.toString(true));
         }
     }
 
+    @Override
+    public boolean getEnableHyperscanFallback() {
+        return Boolean.parseBoolean(sessionVariables.getOrDefault(
+                SessionVariable.ENABLE_HYPERSCAN_FALLBACK, Boolean.toString(true)));
+    }
+
     /**
      * MultiLoadJob will use this constructor
      */
@@ -349,6 +358,8 @@ public RoutineLoadJob(Long id, String name,
         if (ConnectContext.get() != null) {
             SessionVariable var = ConnectContext.get().getSessionVariable();
             sessionVariables.put(SessionVariable.SQL_MODE, Long.toString(var.getSqlMode()));
+            sessionVariables.put(SessionVariable.ENABLE_HYPERSCAN_FALLBACK,
+                    Boolean.toString(var.enableHyperscanFallback));
             this.memtableOnSinkNode = ConnectContext.get().getSessionVariable().enableMemtableOnSinkNode;
             try {
                 this.cloudCluster = ConnectContext.get().getCloudCluster();
@@ -357,6 +368,7 @@ public RoutineLoadJob(Long id, String name,
             }
         } else {
             sessionVariables.put(SessionVariable.SQL_MODE, String.valueOf(SqlModeHelper.MODE_DEFAULT));
+            sessionVariables.put(SessionVariable.ENABLE_HYPERSCAN_FALLBACK, Boolean.toString(true));
         }
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
index 885021440351d7..ec5e3d5a7df57a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
@@ -57,6 +57,7 @@
 import org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
 import org.apache.doris.persist.AlterRoutineLoadJobOperationLog;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.rpc.RpcException;
 import org.apache.doris.service.FrontendOptions;
 import org.apache.doris.thrift.TFileCompressType;
@@ -1120,6 +1121,8 @@ public NereidsRoutineLoadTaskInfo toNereidsRoutineLoadTaskInfo() throws UserExce
         return new NereidsRoutineLoadTaskInfo(execMemLimit, new HashMap<>(jobProperties), maxBatchIntervalS,
                 partitionNamesInfo, mergeType, deleteCondition, sequenceCol, maxFilterRatio, importColumnDescs,
                 precedingFilter, whereExpr, columnSeparator, lineDelimiter, enclose, escape, sendBatchParallelism,
-                loadToSingleTablet, uniqueKeyUpdateMode, partialUpdateNewKeyPolicy, memtableOnSinkNode);
+                loadToSingleTablet, uniqueKeyUpdateMode, partialUpdateNewKeyPolicy, memtableOnSinkNode,
+                Boolean.parseBoolean(sessionVariables.getOrDefault(
+                        SessionVariable.ENABLE_HYPERSCAN_FALLBACK, Boolean.toString(true))));
     }
 }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
index 7cebc3f5165b49..797a7da2eea62e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
@@ -50,6 +50,7 @@
 import org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
 import org.apache.doris.persist.AlterRoutineLoadJobOperationLog;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.thrift.TFileCompressType;
 import org.apache.doris.thrift.TPartialUpdateNewRowPolicy;
 import org.apache.doris.transaction.TransactionState;
@@ -876,7 +877,9 @@ public NereidsRoutineLoadTaskInfo toNereidsRoutineLoadTaskInfo() throws UserExce
         return new NereidsRoutineLoadTaskInfo(execMemLimit, new HashMap<>(jobProperties), maxBatchIntervalS,
                 partitionNamesInfo, mergeType, deleteCondition, sequenceCol, maxFilterRatio, importColumnDescs,
                 precedingFilter, whereExpr, columnSeparator, lineDelimiter, enclose, escape, sendBatchParallelism,
-                loadToSingleTablet, uniqueKeyUpdateMode, partialUpdateNewKeyPolicy, memtableOnSinkNode);
+                loadToSingleTablet, uniqueKeyUpdateMode, partialUpdateNewKeyPolicy, memtableOnSinkNode,
+                Boolean.parseBoolean(sessionVariables.getOrDefault(
+                        SessionVariable.ENABLE_HYPERSCAN_FALLBACK, Boolean.toString(true))));
     }
 
     @Override
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsBrokerLoadTask.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsBrokerLoadTask.java
index 3c05d6678c3148..a721d50ab6ebc7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsBrokerLoadTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsBrokerLoadTask.java
@@ -37,6 +37,7 @@ public class NereidsBrokerLoadTask implements NereidsLoadTaskInfo {
     private boolean strictMode;
     private boolean memtableOnSinkNode;
     private boolean loadToSingleTablet;
+    private boolean enableHyperscanFallback;
     private PartitionNamesInfo partitionNamesInfo;
 
     /**
@@ -44,7 +45,7 @@ public class NereidsBrokerLoadTask implements NereidsLoadTaskInfo {
      */
     public NereidsBrokerLoadTask(long txnId, int timeout, int sendBatchParallelism,
             boolean strictMode, boolean memtableOnSinkNode, boolean loadToSingleTablet,
-            PartitionNamesInfo partitions) {
+            PartitionNamesInfo partitions, boolean enableHyperscanFallback) {
         this.txnId = txnId;
         this.timeout = timeout;
         this.sendBatchParallelism = sendBatchParallelism;
@@ -52,6 +53,7 @@ public NereidsBrokerLoadTask(long txnId, int timeout, int sendBatchParallelism,
         this.memtableOnSinkNode = memtableOnSinkNode;
         this.loadToSingleTablet = loadToSingleTablet;
         this.partitionNamesInfo = partitions;
+        this.enableHyperscanFallback = enableHyperscanFallback;
     }
 
     @Override
@@ -179,6 +181,11 @@ public boolean isLoadToSingleTablet() {
         return loadToSingleTablet;
     }
 
+    @Override
+    public boolean getEnableHyperscanFallback() {
+        return enableHyperscanFallback;
+    }
+
     @Override
     public String getHeaderType() {
         return null;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadTaskInfo.java
index 2ece54c823ecfb..d2c59fc66462e0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadTaskInfo.java
@@ -151,6 +151,8 @@ default int getStreamPerNode() {
         return 2;
     }
 
+    boolean getEnableHyperscanFallback();
+
     /**
      * NereidsImportColumnDescs
      */
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadingTaskPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadingTaskPlanner.java
index 924cc43c1d8e75..000bc87fde5564 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadingTaskPlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadingTaskPlanner.java
@@ -76,6 +76,7 @@ public class NereidsLoadingTaskPlanner {
     private final int sendBatchParallelism;
     private final boolean singleTabletLoadPerSink;
     private final boolean enableMemtableOnSinkNode;
+    private final boolean enableHyperscanFallback;
     private UserIdentity userInfo;
     private final DescriptorTable descTable = new DescriptorTable();
 
@@ -91,7 +92,8 @@ public NereidsLoadingTaskPlanner(Long loadJobId, long txnId, long dbId, OlapTabl
             boolean strictMode, boolean isPartialUpdate, TPartialUpdateNewRowPolicy partialUpdateNewKeyPolicy,
             String timezone, long timeoutS, int loadParallelism,
             int sendBatchParallelism, UserIdentity userInfo,
-            boolean singleTabletLoadPerSink, boolean enableMemtableOnSinkNode) {
+            boolean singleTabletLoadPerSink, boolean enableMemtableOnSinkNode,
+            boolean enableHyperscanFallback) {
         this.loadJobId = loadJobId;
         this.txnId = txnId;
         this.dbId = dbId;
@@ -108,6 +110,7 @@ public NereidsLoadingTaskPlanner(Long loadJobId, long txnId, long dbId, OlapTabl
         this.userInfo = userInfo;
         this.singleTabletLoadPerSink = singleTabletLoadPerSink;
         this.enableMemtableOnSinkNode = enableMemtableOnSinkNode;
+        this.enableHyperscanFallback = enableHyperscanFallback;
     }
 
     /**
@@ -157,7 +160,8 @@ public void plan(TUniqueId loadId, List<List<TBrokerFileStatus>> fileStatusesLis
         }
         NereidsBrokerLoadTask nereidsBrokerLoadTask = new NereidsBrokerLoadTask(txnId, (int) txnTimeout,
                 sendBatchParallelism,
-                strictMode, enableMemtableOnSinkNode, singleTabletLoadPerSink, partitionNamesInfo);
+                strictMode, enableMemtableOnSinkNode, singleTabletLoadPerSink, partitionNamesInfo,
+                enableHyperscanFallback);
 
         TupleDescriptor scanTupleDesc = descTable.createTupleDescriptor();
         scanTupleDesc.setTable(table);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java
index ef159cfb6f494a..822759c047fae2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java
@@ -69,6 +69,7 @@ public class NereidsRoutineLoadTaskInfo implements NereidsLoadTaskInfo {
     protected TUniqueKeyUpdateMode uniquekeyUpdateMode = TUniqueKeyUpdateMode.UPSERT;
     protected TPartialUpdateNewRowPolicy partialUpdateNewKeyPolicy = TPartialUpdateNewRowPolicy.APPEND;
     protected boolean memtableOnSinkNode;
+    protected boolean enableHyperscanFallback;
     protected int timeoutSec;
 
     /**
@@ -80,7 +81,8 @@ public NereidsRoutineLoadTaskInfo(long execMemLimit, Map<String, String> jobProp
             Expression precedingFilter, Expression whereExpr, Separator columnSeparator,
             Separator lineDelimiter, byte enclose, byte escape, int sendBatchParallelism,
             boolean loadToSingleTablet, TUniqueKeyUpdateMode uniqueKeyUpdateMode,
-            TPartialUpdateNewRowPolicy partialUpdateNewKeyPolicy, boolean memtableOnSinkNode) {
+            TPartialUpdateNewRowPolicy partialUpdateNewKeyPolicy, boolean memtableOnSinkNode,
+            boolean enableHyperscanFallback) {
         this.execMemLimit = execMemLimit;
         this.jobProperties = jobProperties;
         this.maxBatchIntervalS = maxBatchIntervalS;
@@ -101,9 +103,15 @@ public NereidsRoutineLoadTaskInfo(long execMemLimit, Map<String, String> jobProp
         this.uniquekeyUpdateMode = uniqueKeyUpdateMode;
         this.partialUpdateNewKeyPolicy = partialUpdateNewKeyPolicy;
         this.memtableOnSinkNode = memtableOnSinkNode;
+        this.enableHyperscanFallback = enableHyperscanFallback;
         this.timeoutSec = calTimeoutSec();
     }
 
+    @Override
+    public boolean getEnableHyperscanFallback() {
+        return enableHyperscanFallback;
+    }
+
     @Override
     public boolean getNegative() {
         return false;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
index b423874e773333..e28a21ddbf56e4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
@@ -313,6 +313,7 @@ public TPipelineFragmentParams plan(TUniqueId loadId, int fragmentInstanceIdInde
         queryOptions.setMemLimit(taskInfo.getMemLimit());
         // for stream load, we use exec_mem_limit to limit the memory usage of load channel.
         queryOptions.setLoadMemLimit(taskInfo.getMemLimit());
+        queryOptions.setEnableHyperscanFallback(taskInfo.getEnableHyperscanFallback());
         // load
         queryOptions.setBeExecVersion(Config.be_exec_version);
         queryOptions.setIsReportSuccess(taskInfo.getEnableProfile() || Config.enable_stream_load_profile);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
index f5ddca41f19a29..b7133704104d5c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
@@ -99,6 +99,7 @@ public class NereidsStreamLoadTask implements NereidsLoadTaskInfo {
     private String groupCommit;
 
     private boolean emptyFieldAsNull = false;
+    private boolean enableHyperscanFallback = true;
 
     /**
      * NereidsStreamLoadTask
@@ -352,18 +353,25 @@ public void setEmptyFieldAsNull(boolean emptyFieldAsNull) {
     /**
      * fromTStreamLoadPutRequest
      */
-    public static NereidsStreamLoadTask fromTStreamLoadPutRequest(TStreamLoadPutRequest request) throws UserException {
+    public static NereidsStreamLoadTask fromTStreamLoadPutRequest(TStreamLoadPutRequest request,
+            boolean enableHyperscanFallback) throws UserException {
         NereidsStreamLoadTask streamLoadTask = new NereidsStreamLoadTask(request.getLoadId(), request.getTxnId(),
                 request.getFileType(), request.getFormatType(),
                 request.getCompressType());
         streamLoadTask.setOptionalFromTSLPutRequest(request);
         streamLoadTask.setGroupCommit(request.getGroupCommitMode());
+        streamLoadTask.enableHyperscanFallback = enableHyperscanFallback;
         if (request.isSetFileSize()) {
             streamLoadTask.fileSize = request.getFileSize();
         }
         return streamLoadTask;
     }
 
+    @Override
+    public boolean getEnableHyperscanFallback() {
+        return enableHyperscanFallback;
+    }
+
     /**
      * setMultiTableBaseTaskInfo
      */
@@ -385,6 +393,7 @@ public void setMultiTableBaseTaskInfo(LoadTaskInfo task) throws UserException {
         this.jsonRoot = task.getJsonRoot();
         this.sendBatchParallelism = task.getSendBatchParallelism();
         this.loadToSingleTablet = task.isLoadToSingleTablet();
+        this.enableHyperscanFallback = task.getEnableHyperscanFallback();
     }
 
     private void setOptionalFromTSLPutRequest(TStreamLoadPutRequest request) throws UserException {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java
index 1a9874056711d7..52ec6efb7b552f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/GroupCommitPlanner.java
@@ -123,7 +123,8 @@ public GroupCommitPlanner(Database db, OlapTable table, List<String> targetColum
                 .setMergeType(TMergeType.APPEND).setThriftRpcTimeoutMs(5000).setLoadId(queryId)
                 .setTrimDoubleQuotes(true).setGroupCommitMode(groupCommit)
                 .setStrictMode(ConnectContext.get().getSessionVariable().enableInsertStrict);
-        NereidsStreamLoadTask streamLoadTask = NereidsStreamLoadTask.fromTStreamLoadPutRequest(streamLoadPutRequest);
+        NereidsStreamLoadTask streamLoadTask = NereidsStreamLoadTask.fromTStreamLoadPutRequest(
+                streamLoadPutRequest, ConnectContext.get().getSessionVariable().enableHyperscanFallback);
         NereidsStreamLoadPlanner planner = new NereidsStreamLoadPlanner(db, table, streamLoadTask);
         // Will using load id as query id in fragment
         // TODO support pipeline
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index 06006086d4dfaa..409df870c989d3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -388,7 +388,8 @@ public Coordinator(ConnectContext context, Planner planner) {
     // Used for broker load task/export task/update coordinator
     // Constructor of Coordinator is too complicated.
     public Coordinator(Long jobId, TUniqueId queryId, DescriptorTable descTable, List<PlanFragment> fragments,
-            List<ScanNode> scanNodes, String timezone, boolean loadZeroTolerance, boolean enableProfile) {
+            List<ScanNode> scanNodes, String timezone, boolean loadZeroTolerance, boolean enableProfile,
+            boolean enableHyperscanFallback) {
         this.jobId = jobId;
         this.queryId = queryId;
         this.descTable = DescriptorToThriftConverter.toThrift(descTable);
@@ -397,6 +398,7 @@ public Coordinator(Long jobId, TUniqueId queryId, DescriptorTable descTable, Lis
         this.queryOptions = new TQueryOptions();
         this.queryOptions.setEnableProfile(enableProfile);
         this.queryOptions.setProfileLevel(2);
+        this.queryOptions.setEnableHyperscanFallback(enableHyperscanFallback);
         this.queryGlobals.setNowString(TimeUtils.getDatetimeFormatWithTimeZone().format(LocalDateTime.now()));
         this.queryGlobals.setTimestampMs(System.currentTimeMillis());
         this.queryGlobals.setTimeZone(timezone);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/CoordinatorContext.java b/fe/fe-core/src/main/java/org/apache/doris/qe/CoordinatorContext.java
index f619d7f4917cc2..d7306bad4de968 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/CoordinatorContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/CoordinatorContext.java
@@ -303,10 +303,11 @@ public static CoordinatorContext buildForLoad(
             List<ScanNode> scanNodes,
             DescriptorTable descTable,
             String timezone, boolean loadZeroTolerance,
-            boolean enableProfile) {
+            boolean enableProfile, boolean enableHyperscanFallback) {
         TQueryOptions queryOptions = new TQueryOptions();
         queryOptions.setEnableProfile(enableProfile);
         queryOptions.setProfileLevel(2);
+        queryOptions.setEnableHyperscanFallback(enableHyperscanFallback);
         queryOptions.setBeExecVersion(Config.be_exec_version);
         queryOptions.setNewVersionUnixTimestamp(true);
         queryOptions.setNewVersionPercentile(true);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/InsertStreamTxnExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/InsertStreamTxnExecutor.java
index 30176665f287c4..291ce981788480 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/InsertStreamTxnExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/InsertStreamTxnExecutor.java
@@ -66,7 +66,8 @@ public void beginTransaction(TStreamLoadPutRequest request) throws UserException
         TTxnParams txnConf = txnEntry.getTxnConf();
         OlapTable table = (OlapTable) txnEntry.getTable();
         // StreamLoadTask's id == request's load_id
-        NereidsStreamLoadTask streamLoadTask = NereidsStreamLoadTask.fromTStreamLoadPutRequest(request);
+        NereidsStreamLoadTask streamLoadTask = NereidsStreamLoadTask.fromTStreamLoadPutRequest(
+                request, ConnectContext.get().getSessionVariable().enableHyperscanFallback);
         NereidsStreamLoadPlanner planner = new NereidsStreamLoadPlanner((Database) txnEntry.getDb(), table,
                 streamLoadTask);
         boolean isMowTable = ((OlapTable) txnEntry.getTable()).getEnableUniqueKeyMergeOnWrite();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
index 35bc335e30468f..bcbd92fad4456d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/NereidsCoordinator.java
@@ -125,11 +125,12 @@ public NereidsCoordinator(ConnectContext context,
     public NereidsCoordinator(Long jobId, TUniqueId queryId, DescriptorTable descTable,
             List<PlanFragment> fragments, List<PipelineDistributedPlan> distributedPlans,
             List<ScanNode> scanNodes, String timezone, boolean loadZeroTolerance,
-            boolean enableProfile) {
-        super(jobId, queryId, descTable, fragments, scanNodes, timezone, loadZeroTolerance, enableProfile);
+            boolean enableProfile, boolean enableHyperscanFallback) {
+        super(jobId, queryId, descTable, fragments, scanNodes, timezone, loadZeroTolerance, enableProfile,
+                enableHyperscanFallback);
         this.coordinatorContext = CoordinatorContext.buildForLoad(
                 this, jobId, queryId, fragments, distributedPlans, scanNodes,
-                descTable, timezone, loadZeroTolerance, enableProfile
+                descTable, timezone, loadZeroTolerance, enableProfile, enableHyperscanFallback
         );
         // same reason in `setForInsert`
         this.coordinatorContext.queryOptions.setDisableFileCache(true);
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 a1592de2947489..a1c1ff0fe8a92c 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
@@ -822,6 +822,8 @@ public String toString() {
 
     public static final String ENABLE_EXTENDED_REGEX = "enable_extended_regex";
 
+    public static final String ENABLE_HYPERSCAN_FALLBACK = "enable_hyperscan_fallback";
+
     public static final String CLOUD_PARTITIONS_TABLE_USE_CACHED_VISIBLE_VERSION =
             "cloud_partitions_table_use_cached_visible_version";
 
@@ -3513,6 +3515,10 @@ public void checkAnnIndexCandidateRowsPercentThreshold(String value) {
             description = "Enable extended regular expressions, support look-around zero-width assertions")
     public boolean enableExtendedRegex = false;
 
+    @VarAttrDef.VarAttr(name = ENABLE_HYPERSCAN_FALLBACK, needForward = true, affectQueryResultInExecution = true,
+            description = "Whether to fall back to RE2 when Hyperscan cannot compile a regular expression")
+    public boolean enableHyperscanFallback = true;
+
     @VarAttrDef.VarAttr(
             name = DEFAULT_VARIANT_SPARSE_HASH_SHARD_COUNT,
             needForward = true,
@@ -5594,6 +5600,7 @@ public TQueryOptions toThrift() {
         tResult.setAnnIndexCandidateRowsPercentThreshold(annIndexCandidateRowsPercentThreshold);
         tResult.setMergeReadSliceSize(mergeReadSliceSizeBytes);
         tResult.setEnableExtendedRegex(enableExtendedRegex);
+        tResult.setEnableHyperscanFallback(enableHyperscanFallback);
         if (fileCacheQueryLimitPercent > 0) {
             tResult.setFileCacheQueryLimitPercent(Math.min(fileCacheQueryLimitPercent,
                     Config.file_cache_query_limit_max_percent));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
index dc898c5a7fcebc..8f7808cf56059e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
@@ -144,6 +144,10 @@ default int getStreamPerNode() {
         return 2;
     }
 
+    default boolean getEnableHyperscanFallback() {
+        return true;
+    }
+
     class ImportColumnDescs {
         @SerializedName("des")
         public List<ImportColumnDesc> descs = Lists.newArrayList();
diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudEnvFactoryTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudEnvFactoryTest.java
index c7af444c99ccb8..de3a20b6a1f7d9 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudEnvFactoryTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudEnvFactoryTest.java
@@ -68,7 +68,7 @@ public void testLegacyLoadCoordinatorSetsFunctionVersionOptions() {
             context.setThreadLocalInfo();
             Coordinator coordinator = new CloudEnvFactory().createCoordinator(
                     1L, new TUniqueId(1L, 1L), new DescriptorTable(),
-                    Collections.emptyList(), Collections.emptyList(), "UTC", false, false);
+                    Collections.emptyList(), Collections.emptyList(), "UTC", false, false, true);
 
             Assert.assertTrue(coordinator instanceof CloudCoordinator);
             Assert.assertTrue(coordinator.getQueryOptions().isSetNewVersionUnixTimestamp());
diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
index 8f266e81ab82e7..465127c2ce381b 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java
@@ -19,6 +19,7 @@
 
 import org.apache.doris.analysis.UserIdentity;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.OlapTable;
 import org.apache.doris.cloud.catalog.CloudEnv;
 import org.apache.doris.cloud.system.CloudSystemInfoService;
 import org.apache.doris.common.Config;
@@ -26,7 +27,9 @@
 import org.apache.doris.common.jmockit.Deencapsulation;
 import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.mysql.privilege.Auth;
+import org.apache.doris.nereids.load.NereidsStreamLoadTask;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.SystemInfoService;
 import org.apache.doris.thrift.TStreamLoadPutRequest;
@@ -39,6 +42,7 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 public class StreamLoadHandlerTest {
     @Test
@@ -145,6 +149,38 @@ public void testGroupCommitValidatesBackendComputeGroupPrivilege() throws Except
         }
     }
 
+    @Test
+    public void testGeneratePlanUsesSessionVariableSnapshot() throws Exception {
+        TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+        OlapTable table = Mockito.mock(OlapTable.class);
+        Mockito.when(table.tryReadLock(Mockito.anyLong(), Mockito.any(TimeUnit.class))).thenReturn(true);
+
+        SessionVariable sessionVariable = new SessionVariable();
+        sessionVariable.enableHyperscanFallback = false;
+        ConnectContext context = new ConnectContext();
+        context.setSessionVariable(sessionVariable);
+        context.setThreadLocalInfo();
+
+        try (MockedStatic<NereidsStreamLoadTask> mockedTask = Mockito.mockStatic(NereidsStreamLoadTask.class)) {
+            mockedTask.when(() -> NereidsStreamLoadTask.fromTStreamLoadPutRequest(request, false))
+                    .thenThrow(new DdlException("stop after checking session variable"));
+
+            StreamLoadHandler handler = new StreamLoadHandler(
+                    request, null, new TStreamLoadPutResult(), "127.0.0.1");
+            try {
+                handler.generatePlan(table);
+                Assert.fail("generatePlan should use the session variable snapshot");
+            } catch (DdlException e) {
+                Assert.assertTrue(e.getMessage().contains("stop after checking session variable"));
+            }
+
+            mockedTask.verify(() -> NereidsStreamLoadTask.fromTStreamLoadPutRequest(request, false));
+            Mockito.verify(table).readUnlock();
+        } finally {
+            ConnectContext.remove();
+        }
+    }
+
     private Backend createBackend(long id, String host) {
         Backend backend = new Backend(id, host, 9050);
         backend.setAlive(true);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
index d2724063abfab2..61ddb440e8edcf 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
@@ -34,6 +34,7 @@
 import org.apache.doris.load.routineload.kafka.KafkaTaskInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
 import org.apache.doris.persist.EditLog;
+import org.apache.doris.qe.SessionVariable;
 import org.apache.doris.thrift.TKafkaRLTaskProgress;
 import org.apache.doris.thrift.TLoadSourceType;
 import org.apache.doris.thrift.TRLTaskTxnCommitAttachment;
@@ -58,6 +59,18 @@
 import java.util.Map;
 
 public class RoutineLoadJobTest {
+
+    @Test
+    public void testHyperscanFallbackReplayCompatibility() {
+        KafkaRoutineLoadJob job = new KafkaRoutineLoadJob();
+        Map<String, String> sessionVariables = Maps.newHashMap();
+        Deencapsulation.setField(job, "sessionVariables", sessionVariables);
+        Assert.assertTrue(job.getEnableHyperscanFallback());
+
+        sessionVariables.put(SessionVariable.ENABLE_HYPERSCAN_FALLBACK, Boolean.toString(false));
+        Assert.assertFalse(job.getEnableHyperscanFallback());
+    }
+
     @Test
     public void testFirstErrorMsgInTxnCommitAttachment() {
         String overlongFirstErrorMsg = Strings.repeat("x", Config.first_error_msg_max_length + 1);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsLoadScanProviderTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsLoadScanProviderTest.java
index 95d469a4a93896..560dbbff5a8eab 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsLoadScanProviderTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsLoadScanProviderTest.java
@@ -60,13 +60,26 @@ public void testArrowStreamLoadKeepsExplicitColumnCase() throws Exception {
         request.setCompressType(TFileCompressType.UNKNOWN);
         request.setColumns("time,securityid,EV");
 
-        NereidsStreamLoadTask task = NereidsStreamLoadTask.fromTStreamLoadPutRequest(request);
+        NereidsStreamLoadTask task = NereidsStreamLoadTask.fromTStreamLoadPutRequest(request, true);
         NereidsDataDescription dataDescription = new NereidsDataDescription("t_upper", task);
 
         Assertions.assertEquals(Lists.newArrayList("time", "securityid", "EV"),
                 dataDescription.getFileFieldNames());
     }
 
+    @Test
+    public void testStreamLoadPreservesHyperscanFallbackOption() throws Exception {
+        TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+        request.setLoadId(new TUniqueId(1, 2));
+        request.setTxnId(3);
+        request.setFileType(TFileType.FILE_STREAM);
+        request.setFormatType(TFileFormatType.FORMAT_CSV_PLAIN);
+        request.setCompressType(TFileCompressType.PLAIN);
+
+        NereidsStreamLoadTask task = NereidsStreamLoadTask.fromTStreamLoadPutRequest(request, false);
+        Assertions.assertFalse(task.getEnableHyperscanFallback());
+    }
+
     @Test
     public void testArrowBrokerLoadKeepsExplicitColumnCaseForNonLowercaseFormat() {
         NereidsDataDescription dataDescription = new NereidsDataDescription("t_upper", null,
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/load/VariantLoadParseInjectionTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/VariantLoadParseInjectionTest.java
index b5beafa169356b..23de8e7852a815 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/load/VariantLoadParseInjectionTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/VariantLoadParseInjectionTest.java
@@ -160,7 +160,7 @@ private LoadFixture createRoutineFixture(InputShape shape) throws Exception {
         NereidsRoutineLoadTaskInfo task = new NereidsRoutineLoadTaskInfo(1024L, new HashMap<>(), 10L,
                 null, LoadTask.MergeType.APPEND, null, null, 0.0, columnDescs(shape), null, null,
                 null, null, (byte) 0, (byte) 0, 1, false, TUniqueKeyUpdateMode.UPSERT,
-                TPartialUpdateNewRowPolicy.APPEND, false);
+                TPartialUpdateNewRowPolicy.APPEND, false, true);
         return createStreamFamilyFixture("routine/" + shape, task, new TUniqueId(4L, 5L));
     }
 
diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorTest.java
index 1cdc3163d909f1..5852f677f8cc7d 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorTest.java
@@ -141,7 +141,7 @@ public void testFragmentExecParamsMarksNonOlapTopnFilterSource() {
 
         Coordinator.FragmentExecParams fragParams = new Coordinator(0L, new TUniqueId(1L, 1L),
                 new DescriptorTable(), Collections.singletonList(fragment), Collections.singletonList(scanNode),
-                "UTC", false, false).new FragmentExecParams(fragment);
+                "UTC", false, false, true).new FragmentExecParams(fragment);
         TNetworkAddress host = new TNetworkAddress("127.0.0.1", 9060);
         fragParams.instanceExecParams.add(
                 new Coordinator.FInstanceExecParam(new TUniqueId(2L, 2L), host, fragParams));
@@ -151,6 +151,13 @@ public void testFragmentExecParamsMarksNonOlapTopnFilterSource() {
         Mockito.verify(sortNode).setHasRuntimePredicate();
     }
 
+    @Test
+    public void testBrokerLoadPreservesHyperscanFallbackOption() {
+        Coordinator coordinator = new Coordinator(0L, new TUniqueId(1L, 1L), new DescriptorTable(),
+                Collections.emptyList(), Collections.emptyList(), "UTC", false, false, false);
+        Assertions.assertFalse(coordinator.getQueryOptions().isEnableHyperscanFallback());
+    }
+
     private NereidsPlanner plan(String sql) throws IOException {
         connectContext.getSessionVariable().setDisableNereidsRules(
                 "PRUNE_EMPTY_PARTITION,OLAP_SCAN_TABLET_PRUNE");
diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
index 9febc45b150e6d..b7f7b8807c4e24 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/SessionVariablesTest.java
@@ -408,4 +408,15 @@ public void testCoordinatorThriftLimitPropagatesToBackends() {
                 queryOptions.getCoordinatorThriftMaxMessageSize());
         Assertions.assertTrue(queryOptions.isSupportsExternalFileReportAck());
     }
+
+    @Test
+    public void testHyperscanFallbackPropagatesToBackends() throws Exception {
+        SessionVariable variable = new SessionVariable();
+        Assertions.assertTrue(variable.toThrift().isEnableHyperscanFallback());
+
+        VariableMgr.setVar(variable, new SetVar(SetType.SESSION,
+                SessionVariable.ENABLE_HYPERSCAN_FALLBACK, new StringLiteral("false")));
+
+        Assertions.assertFalse(variable.toThrift().isEnableHyperscanFallback());
+    }
 }
diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift
index 1f05f5f2312fc4..e592459f77f24b 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -516,6 +516,8 @@ struct TQueryOptions {
   229: optional i32 coordinator_thrift_max_message_size;
   // FE can explicitly and idempotently acknowledge external-file commit reports.
   230: optional bool supports_external_file_report_ack = false;
+  // Fall back to RE2 when Hyperscan cannot compile a regular expression.
+  231: optional bool enable_hyperscan_fallback = true;
   // For cloud, to control if the content would be written into file cache
   // In write path, to control if the content would be written into file cache.
   // In read path, read from file cache or remote storage when execute query.