From 34ae8b8744b8b9f469c36653d3fb5b002f1e876d Mon Sep 17 00:00:00 2001 From: happenlee Date: Fri, 14 Aug 2026 18:19:06 +0800 Subject: [PATCH 1/6] [fix](be) Fall back from expensive Hyperscan bounded repeats ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Hyperscan compilation can become extremely expensive for regular expressions with large bounded repetitions, such as `prompt_rewrite\.h03.{0,1000}429`. Doris previously sent every compatible pattern to Hyperscan before considering RE2, so compiling such expressions could consume excessive CPU and delay query execution. Detect bounded repetitions above 50 before calling Hyperscan and reuse the existing RE2 fallback path for both constant and non-constant patterns. Keep the detector local to the LIKE/REGEXP implementation and cover its threshold behavior and end-to-end matching results. ### Release note Fall back to RE2 for regular expressions whose bounded repetition exceeds 50 to avoid expensive Hyperscan compilation. ### Check List (For Author) - Test: Unit Test - `./run-be-ut.sh -j 48 --run --filter=FunctionLikeTest.hyperscan_bounded_repeat_fallback:FunctionLikeTest.hyperscan_bounded_repeat_threshold` - Behavior changed: Yes. Large bounded repetitions use RE2 instead of Hyperscan while preserving REGEXP results. - Does this need documentation: No --- be/src/exprs/function/like.cpp | 78 +++++++++++++++++++ be/src/exprs/function/like.h | 3 + be/test/exprs/function/function_like_test.cpp | 30 +++++++ 3 files changed, 111 insertions(+) diff --git a/be/src/exprs/function/like.cpp b/be/src/exprs/function/like.cpp index 8bbfcf9b81dada..0c1f94cf0f2228 100644 --- a/be/src/exprs/function/like.cpp +++ b/be/src/exprs/function/like.cpp @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -36,6 +37,70 @@ #include "exprs/function/simple_function_factory.h" 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; +} + +/// Bounded repetitions can expand Hyperscan's compiler graph and make compilation extremely +/// expensive. This checker is adapted from ClickHouse's `SlowWithHyperscanChecker`. +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 { + return is_slow_one_repeat(regexp) || is_slow_two_repeats(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 + // A regex to match any regex pattern is equivalent to a substring search. static const RE2 SUBSTRING_RE(R"((?:\.\*)*([^\.\^\{\[\(\|\)\]\}\+\*\?\$\\]*)(?:\.\*)*)"); @@ -487,8 +552,21 @@ 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) { + static const SlowWithHyperscanChecker slow_with_hyperscan_checker; + return slow_with_hyperscan_checker.is_slow(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; + // Do not call FunctionContext::set_error here, since callers fall back to RE2. + return Status::RuntimeError( + "Skip hyperscan compilation because bounded repetition exceeds 50"); + } + 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); diff --git a/be/src/exprs/function/like.h b/be/src/exprs/function/like.h index 461c97956bcc7f..f2babf4ff636f3 100644 --- a/be/src/exprs/function/like.h +++ b/be/src/exprs/function/like.h @@ -29,6 +29,7 @@ #include #include #include +#include #include "common/status.h" #include "core/block/column_numbers.h" @@ -292,6 +293,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/test/exprs/function/function_like_test.cpp b/be/test/exprs/function/function_like_test.cpp index 82618a790e99e7..bf0b139062df58 100644 --- a/be/test/exprs/function/function_like_test.cpp +++ b/be/test/exprs/function/function_like_test.cpp @@ -32,6 +32,11 @@ namespace doris { +class FunctionLikeTestHelper : public FunctionLikeBase { +public: + using FunctionLikeBase::should_fallback_to_re2; +}; + TEST(FunctionLikeTest, like) { std::string func_name = "like"; @@ -137,6 +142,31 @@ 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(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 }")); +} + TEST(FunctionLikeTest, regexp_extract) { std::string func_name = "regexp_extract"; From f34296eb897166f65da395180e2a34ad6570eb4f Mon Sep 17 00:00:00 2001 From: happenlee Date: Fri, 14 Aug 2026 22:44:05 +0800 Subject: [PATCH 2/6] [fix](be) Add Hyperscan fallback session control ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Hyperscan compilation can fail or be intentionally intercepted for regular expressions with expensive bounded repetitions. Doris always fell back to RE2, so users could not choose strict failure behavior. Add the enable_hyperscan_fallback session variable, propagate it through TQueryOptions, and return the Hyperscan status when fallback is disabled. Mask escaped characters and character classes before bounded-repeat detection so literal braces are not intercepted. ### Release note Add the enable_hyperscan_fallback session variable. It defaults to true; setting it to false returns an error instead of falling back to RE2 when Hyperscan compilation is unavailable. ### Check List (For Author) - Test: - Unit Test: FunctionLikeTest.* and org.apache.doris.qe.SessionVariablesTest - Behavior changed: Yes. Hyperscan fallback can now be disabled per session; the default behavior is unchanged. - Does this need documentation: No --- be/src/exprs/function/like.cpp | 75 +++++++++++++++++-- be/src/exprs/function/like.h | 4 +- be/test/exprs/function/function_like_test.cpp | 67 +++++++++++++++++ .../org/apache/doris/qe/SessionVariable.java | 7 ++ .../apache/doris/qe/SessionVariablesTest.java | 11 +++ gensrc/thrift/PaloInternalService.thrift | 2 + 6 files changed, 157 insertions(+), 9 deletions(-) diff --git a/be/src/exprs/function/like.cpp b/be/src/exprs/function/like.cpp index 0c1f94cf0f2228..b49f7c373304f1 100644 --- a/be/src/exprs/function/like.cpp +++ b/be/src/exprs/function/like.cpp @@ -45,6 +45,44 @@ bool is_larger_than_fifty(std::string_view str) { 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; + 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; + } + 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_close) { + character_class_can_close = true; + } + continue; + } + if (current == '[') { + masked_character = ' '; + in_character_class = true; + character_class_can_close = false; + } + } + return masked_regexp; +} + /// Bounded repetitions can expand Hyperscan's compiler graph and make compilation extremely /// expensive. This checker is adapted from ClickHouse's `SlowWithHyperscanChecker`. class SlowWithHyperscanChecker { @@ -54,7 +92,8 @@ class SlowWithHyperscanChecker { _searcher_two_repeats(R"(\{\s*([\d]+)\s*,\s*([\d]+)\s*\})") {} bool is_slow(std::string_view regexp) const { - return is_slow_one_repeat(regexp) || is_slow_two_repeats(regexp); + 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: @@ -248,8 +287,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); @@ -517,7 +557,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); @@ -532,6 +573,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); @@ -562,7 +606,7 @@ Status FunctionLikeBase::hs_prepare(FunctionContext* context, const char* expres if (should_fallback_to_re2(expression)) { *database = nullptr; *scratch = nullptr; - // Do not call FunctionContext::set_error here, since callers fall back to RE2. + // Callers either fall back to RE2 or return this status based on the session variable. return Status::RuntimeError( "Skip hyperscan compilation because bounded repetition exceeds 50"); } @@ -575,7 +619,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("hs_compile regex pattern error:" + error_message); } hs_free_compile_error(compile_err); @@ -584,7 +628,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("hs_alloc_scratch allocate scratch space error"); } @@ -1020,12 +1064,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(); @@ -1052,6 +1103,8 @@ Status FunctionLike::open(FunctionContext* context, FunctionContext::FunctionSta } std::shared_ptr state = std::make_shared(); 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)) { @@ -1082,6 +1135,8 @@ Status FunctionRegexpLike::open(FunctionContext* context, std::shared_ptr state = std::make_shared(); 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)) { @@ -1113,12 +1168,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 f2babf4ff636f3..d648919363dce2 100644 --- a/be/src/exprs/function/like.h +++ b/be/src/exprs/function/like.h @@ -183,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 @@ -227,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; diff --git a/be/test/exprs/function/function_like_test.cpp b/be/test/exprs/function/function_like_test.cpp index bf0b139062df58..34bf0ad407a5c9 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 +#include #include +#include +#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,7 +31,9 @@ #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 { @@ -37,6 +43,39 @@ class FunctionLikeTestHelper : public FunctionLikeBase { using FunctionLikeBase::should_fallback_to_re2; }; +template +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(); + auto context = FunctionContext::create_context( + &runtime_state, std::make_shared(), {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> constant_columns(2); + if (constant_known_at_open) { + constant_columns[1] = std::make_shared(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(), "result"}); + return function.execute_impl(context.get(), block, {0, 1}, 2, 1); +} + TEST(FunctionLikeTest, like) { std::string func_name = "like"; @@ -165,6 +204,34 @@ TEST(FunctionLikeTest, hyperscan_bounded_repeat_threshold) { 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}]")); +} + +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( + "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); + } +} + +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( + "a{51}", R"(a\{51\})", constant_known_at_open) + .ok()); + EXPECT_TRUE(execute_pattern_with_fallback_disabled( + "a", "[a{51}]", constant_known_at_open) + .ok()); + EXPECT_TRUE(execute_pattern_with_fallback_disabled("a{51}", "_{51}", + constant_known_at_open) + .ok()); + } } TEST(FunctionLikeTest, regexp_extract) { 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/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. From 17e768572e5cc2d7bd95771f988b63e2dc04c99d Mon Sep 17 00:00:00 2001 From: happenlee Date: Sat, 15 Aug 2026 00:36:36 +0800 Subject: [PATCH 3/6] [fix](be) Handle repeats after negated character classes ### What problem does this PR solve? Issue Number: None Related PR: #66788 Problem Summary: The Hyperscan bounded-repeat masker treated every leading caret in a character class as the negation marker. For `[^^]`, the second caret is class content, but Doris kept the class open and masked the real bounded repeat that followed it. Track whether the one optional leading negation marker is still allowed so the closing bracket is recognized and the expensive repeat is intercepted. ### Release note Correct expensive Hyperscan repeat detection after negated character classes. ### Check List (For Author) - Test: Unit Test - `GLIBC_COMPATIBILITY=OFF ./run-be-ut.sh -j 48 --run --filter=FunctionLikeTest.*` - Behavior changed: Yes. Large bounded repeats following character classes such as `[^^]` are now intercepted instead of reaching Hyperscan. - Does this need documentation: No --- be/src/exprs/function/like.cpp | 8 +++++++- be/test/exprs/function/function_like_test.cpp | 6 ++++++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/be/src/exprs/function/like.cpp b/be/src/exprs/function/like.cpp index b49f7c373304f1..cff1164b4ae4c5 100644 --- a/be/src/exprs/function/like.cpp +++ b/be/src/exprs/function/like.cpp @@ -50,6 +50,7 @@ std::string mask_escaped_characters_and_character_classes(std::string_view regex 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) { @@ -57,6 +58,7 @@ std::string mask_escaped_characters_and_character_classes(std::string_view regex escaped = false; if (in_character_class) { character_class_can_close = true; + character_class_can_negate = false; } continue; } @@ -69,8 +71,11 @@ std::string mask_escaped_characters_and_character_classes(std::string_view regex masked_character = ' '; if (current == ']' && character_class_can_close) { in_character_class = false; - } else if (current != '^' || character_class_can_close) { + } else if (current == '^' && character_class_can_negate) { + character_class_can_negate = false; + } else { character_class_can_close = true; + character_class_can_negate = false; } continue; } @@ -78,6 +83,7 @@ std::string mask_escaped_characters_and_character_classes(std::string_view regex masked_character = ' '; in_character_class = true; character_class_can_close = false; + character_class_can_negate = true; } } return masked_regexp; diff --git a/be/test/exprs/function/function_like_test.cpp b/be/test/exprs/function/function_like_test.cpp index 34bf0ad407a5c9..45f2b3af91dde2 100644 --- a/be/test/exprs/function/function_like_test.cpp +++ b/be/test/exprs/function/function_like_test.cpp @@ -206,6 +206,7 @@ TEST(FunctionLikeTest, hyperscan_bounded_repeat_threshold) { 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) { @@ -216,6 +217,11 @@ TEST(FunctionLikeTest, hyperscan_bounded_repeat_fallback_disabled) { 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( + "^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); } } From eecadf3d2b74572e110522d20202a3148032624e Mon Sep 17 00:00:00 2001 From: happenlee Date: Sat, 15 Aug 2026 02:01:59 +0800 Subject: [PATCH 4/6] [fix](be) Complete Hyperscan bounded-repeat safeguards ### What problem does this PR solve? Issue Number: None Related PR: #66788 Problem Summary: The expensive bounded-repeat check was local to LIKE/REGEXP, so MATCH_REGEXP, inverted-index regexp queries, and multi-match could still compile pathological expressions directly with Hyperscan. Load planners also created fresh query options and lost the enable_hyperscan_fallback value selected when broker, routine, or stream load work was created. Move the checker into a shared utility, reject expensive patterns on Hyperscan-only paths, and persist and propagate the fallback option through load jobs, task descriptors, and both legacy and Nereids coordinators. Old replayed jobs without the persisted value retain the default enabled behavior. ### Release note Apply bounded-repeat safeguards to all Hyperscan compilation paths and preserve enable_hyperscan_fallback for load execution. ### Check List (For Author) - Test: Unit Test - BE: 40 targeted FunctionLike, MATCH_REGEXP, multi-match, and inverted-index regexp tests - FE: 32 targeted load task, coordinator, and routine-load replay tests - Behavior changed: Yes. Hyperscan-only paths reject expensive bounded repeats, while load execution now honors the fallback setting captured when the work was created. - Does this need documentation: No --- be/src/exprs/function/like.cpp | 117 +-------------- be/src/exprs/function/match.cpp | 5 + be/src/exprs/function/regexps.h | 4 + .../index/inverted/query/regexp_query.cpp | 4 + .../query_v2/regexp_query/regexp_weight.cpp | 4 + be/src/util/hyperscan_util.cpp | 139 ++++++++++++++++++ be/src/util/hyperscan_util.h | 29 ++++ .../exprs/function/function_match_test.cpp | 20 ++- .../function/function_multi_match_test.cpp | 6 + .../inverted/query/regexp_query_test.cpp | 23 ++- .../inverted/query_v2/regexp_query_test.cpp | 15 +- .../org/apache/doris/catalog/EnvFactory.java | 8 +- .../doris/cloud/catalog/CloudEnvFactory.java | 8 +- .../doris/cloud/load/CloudBrokerLoadJob.java | 5 +- .../cloud/load/CloudLoadLoadingTask.java | 5 +- .../doris/cloud/qe/CloudCoordinator.java | 5 +- .../apache/doris/load/StreamLoadHandler.java | 4 +- .../doris/load/loadv2/BrokerLoadJob.java | 4 +- .../apache/doris/load/loadv2/BulkLoadJob.java | 3 + .../doris/load/loadv2/LoadLoadingTask.java | 10 +- .../load/routineload/RoutineLoadJob.java | 12 ++ .../kafka/KafkaRoutineLoadJob.java | 5 +- .../kinesis/KinesisRoutineLoadJob.java | 5 +- .../nereids/load/NereidsBrokerLoadTask.java | 9 +- .../nereids/load/NereidsLoadTaskInfo.java | 2 + .../load/NereidsLoadingTaskPlanner.java | 8 +- .../load/NereidsRoutineLoadTaskInfo.java | 10 +- .../load/NereidsStreamLoadPlanner.java | 1 + .../nereids/load/NereidsStreamLoadTask.java | 11 +- .../doris/planner/GroupCommitPlanner.java | 3 +- .../java/org/apache/doris/qe/Coordinator.java | 4 +- .../apache/doris/qe/CoordinatorContext.java | 3 +- .../doris/qe/InsertStreamTxnExecutor.java | 3 +- .../apache/doris/qe/NereidsCoordinator.java | 7 +- .../org/apache/doris/task/LoadTaskInfo.java | 4 + .../cloud/catalog/CloudEnvFactoryTest.java | 2 +- .../load/routineload/RoutineLoadJobTest.java | 12 ++ .../load/NereidsLoadScanProviderTest.java | 15 +- .../load/VariantLoadParseInjectionTest.java | 2 +- .../org/apache/doris/qe/CoordinatorTest.java | 9 +- 40 files changed, 394 insertions(+), 151 deletions(-) create mode 100644 be/src/util/hyperscan_util.cpp create mode 100644 be/src/util/hyperscan_util.h diff --git a/be/src/exprs/function/like.cpp b/be/src/exprs/function/like.cpp index cff1164b4ae4c5..938ecfb958fa1e 100644 --- a/be/src/exprs/function/like.cpp +++ b/be/src/exprs/function/like.cpp @@ -19,9 +19,7 @@ #include #include -#include -#include #include #include #include @@ -35,116 +33,9 @@ #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 { -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; -} - -/// Bounded repetitions can expand Hyperscan's compiler graph and make compilation extremely -/// expensive. This checker is adapted from ClickHouse's `SlowWithHyperscanChecker`. -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 // A regex to match any regex pattern is equivalent to a substring search. static const RE2 SUBSTRING_RE(R"((?:\.\*)*([^\.\^\{\[\(\|\)\]\}\+\*\?\$\\]*)(?:\.\*)*)"); @@ -603,8 +494,7 @@ 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) { - static const SlowWithHyperscanChecker slow_with_hyperscan_checker; - return slow_with_hyperscan_checker.is_slow(regexp); + return is_hyperscan_regexp_expensive(regexp); } Status FunctionLikeBase::hs_prepare(FunctionContext* context, const char* expression, @@ -613,8 +503,7 @@ Status FunctionLikeBase::hs_prepare(FunctionContext* context, const char* expres *database = nullptr; *scratch = nullptr; // Callers either fall back to RE2 or return this status based on the session variable. - return Status::RuntimeError( - "Skip hyperscan compilation because bounded repetition exceeds 50"); + return Status::RuntimeError(HYPERSCAN_BOUNDED_REPEAT_ERROR); } hs_compile_error_t* compile_err; 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(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& 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 +#include + +#include +#include + +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 + +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_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 patterns = {"(ab?c?d){1000,5000}"}; + EXPECT_THROW((multiregexps::constructRegexps(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 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(); + 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 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(); + context->collection_statistics = std::make_shared(); + context->collection_similarity = std::make_shared(); + + std::wstring field = StringHelper::to_wstring("content"); + auto query = std::make_shared(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(); @@ -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 fragments, List 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 fragments, List 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..bbadec95473cca 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 @@ -40,6 +40,7 @@ import org.apache.doris.load.loadv2.JobState; import org.apache.doris.load.loadv2.LoadLoadingTask; import org.apache.doris.load.loadv2.LoadTask; +import org.apache.doris.qe.SessionVariable; import org.apache.doris.qe.AutoCloseConnectContext; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.Coordinator; @@ -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 fragments, List 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..ed8c9de6580378 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 @@ -41,6 +41,7 @@ import org.apache.doris.nereids.load.NereidsStreamLoadTask; import org.apache.doris.planner.GroupCommitPlanner; import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.VariableMgr; import org.apache.doris.service.ExecuteEnv; import org.apache.doris.system.Backend; import org.apache.doris.system.SystemInfoService; @@ -269,7 +270,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, VariableMgr.getDefaultSessionVariable().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 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> fileStatusList, @@ -134,7 +137,8 @@ public void init(TUniqueId loadId, List> 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..e552555508c6b7 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 @@ -53,6 +53,7 @@ import org.apache.doris.nereids.load.NereidsLoadUtils; import org.apache.doris.nereids.load.NereidsRoutineLoadTaskInfo; import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.qe.SessionVariable; import org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand; import org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo; import org.apache.doris.persist.AlterRoutineLoadJobOperationLog; @@ -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> 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 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 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 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 fragments, - List scanNodes, String timezone, boolean loadZeroTolerance, boolean enableProfile) { + List 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 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 fragments, List distributedPlans, List 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/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 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/routineload/RoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java index d2724063abfab2..5bba4635275fb5 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,17 @@ import java.util.Map; public class RoutineLoadJobTest { + + @Test + public void testHyperscanFallbackReplayCompatibility() { + KafkaRoutineLoadJob job = new KafkaRoutineLoadJob(); + Map 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"); From ed8c4c190c1cd61508bdc1d27766e5187feb52e2 Mon Sep 17 00:00:00 2001 From: happenlee Date: Sat, 15 Aug 2026 02:22:10 +0800 Subject: [PATCH 5/6] [fix](fe) Fix load fallback code style ### What problem does this PR solve? Issue Number: None Related PR: #66788 Problem Summary: The load fallback propagation change placed SessionVariable imports out of lexicographical order and omitted the required blank line between two test methods, causing FE checkstyle to fail. Restore the expected import ordering and method separation. ### Release note None ### Check List (For Author) - Test: No need to test (code-style-only change; per request, compilation and tests were not rerun) - Behavior changed: No - Does this need documentation: No --- .../java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java | 2 +- .../doris/load/routineload/kafka/KafkaRoutineLoadJob.java | 2 +- .../org/apache/doris/load/routineload/RoutineLoadJobTest.java | 1 + 3 files changed, 3 insertions(+), 2 deletions(-) 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 bbadec95473cca..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 @@ -40,12 +40,12 @@ import org.apache.doris.load.loadv2.JobState; import org.apache.doris.load.loadv2.LoadLoadingTask; import org.apache.doris.load.loadv2.LoadTask; -import org.apache.doris.qe.SessionVariable; import org.apache.doris.qe.AutoCloseConnectContext; import org.apache.doris.qe.ConnectContext; 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; 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 e552555508c6b7..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 @@ -53,11 +53,11 @@ import org.apache.doris.nereids.load.NereidsLoadUtils; import org.apache.doris.nereids.load.NereidsRoutineLoadTaskInfo; import org.apache.doris.nereids.trees.expressions.Expression; -import org.apache.doris.qe.SessionVariable; import org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand; 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; 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 5bba4635275fb5..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 @@ -70,6 +70,7 @@ public void testHyperscanFallbackReplayCompatibility() { 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); From 68f72081dccb35377e5f4793c5f7841425b88e74 Mon Sep 17 00:00:00 2001 From: happenlee Date: Sun, 16 Aug 2026 12:08:07 +0800 Subject: [PATCH 6/6] [fix](fe) Use session snapshot for stream load planning ### What problem does this PR solve? Issue Number: None Related PR: #66788 Problem Summary: RPC stream load planning read enable_hyperscan_fallback directly from the mutable global default session variable without holding VariableMgr's read lock. Use the request-scoped ConnectContext session snapshot instead so planning observes a consistent value, and add a unit test that distinguishes the snapshot value from the global default. ### Release note None ### Check List (For Author) - Test: Unit Test - ./run-fe-ut.sh --run org.apache.doris.load.StreamLoadHandlerTest - Behavior changed: No - Does this need documentation: No --- .../apache/doris/load/StreamLoadHandler.java | 3 +- .../doris/load/StreamLoadHandlerTest.java | 36 +++++++++++++++++++ 2 files changed, 37 insertions(+), 2 deletions(-) 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 ed8c9de6580378..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 @@ -41,7 +41,6 @@ import org.apache.doris.nereids.load.NereidsStreamLoadTask; import org.apache.doris.planner.GroupCommitPlanner; import org.apache.doris.qe.ConnectContext; -import org.apache.doris.qe.VariableMgr; import org.apache.doris.service.ExecuteEnv; import org.apache.doris.system.Backend; import org.apache.doris.system.SystemInfoService; @@ -271,7 +270,7 @@ public void generatePlan(OlapTable table) throws UserException { } try { NereidsStreamLoadTask streamLoadTask = NereidsStreamLoadTask.fromTStreamLoadPutRequest( - request, VariableMgr.getDefaultSessionVariable().enableHyperscanFallback); + request, ConnectContext.get().getSessionVariable().enableHyperscanFallback); if (isMultiTableRequest) { buildMultiTableStreamLoadTask(streamLoadTask, request.getTxnId()); } 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 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);