[BugFix] Fix query cache crash when enable group execution

[stdpain]

Why I'm doing:

When group execution and query cache are turned on, if a runtime filter is pushed down to AGG instead of SCAN, it will cause the colocate runtime filter to use the wrong runtime filter, resulting in a crash.

I apply this patch to make runtime filter evaluate on aggregate node:

diff --git a/fe/fe-core/src/main/java/com/starrocks/planner/RuntimeFilterDescription.java b/fe/fe-core/src/main/java/com/starrocks/planner/RuntimeFilterDescription.java
index 282f355ad5..352a52cdfd 100644
--- a/fe/fe-core/src/main/java/com/starrocks/planner/RuntimeFilterDescription.java
+++ b/fe/fe-core/src/main/java/com/starrocks/planner/RuntimeFilterDescription.java
@@ -189,6 +189,9 @@ public class RuntimeFilterDescription {
 
     // return true if Node could accept the Filter
     public boolean canAcceptFilter(PlanNode node, RuntimeFilterPushDownContext rfPushCtx) {
+        if (node instanceof ScanNode) {
+            return false;
+        }
         if (RuntimeFilterType.TOPN_FILTER.equals(runtimeFilterType())) {
             if (node instanceof ScanNode) {
                 ScanNode scanNode = (ScanNode) node;

then execution the test case:

select count(*) from (select c0,c1 from t0 group by c0,c1)l join t1 r on l.c1 = r.c1 and l.c0=r.c0 where r.c2 < 5000;

    @         0x11ee3b9c starrocks::failure_function()
    @         0x2119856e google::LogMessage::Fail()
    @         0x211997e9 google::LogMessageFatal::~LogMessageFatal()
    @         0x12447378 starrocks::RuntimeFilterProbeDescriptor::runtime_filter(int) const
    @         0x1a3ff6a3 starrocks::RuntimeFilterProbeCollector::update_selectivity(starrocks::Chunk*, starrocks::RuntimeBloomFilterEvalContext&)
    @         0x1a3fb165 starrocks::RuntimeFilterProbeCollector::do_evaluate(starrocks::Chunk*, starrocks::RuntimeBloomFilterEvalContext&)
    @         0x1a3fdaab starrocks::RuntimeFilterProbeCollector::evaluate(starrocks::Chunk*, starrocks::RuntimeBloomFilterEvalContext&)
    @         0x14975c81 starrocks::pipeline::Operator::eval_runtime_bloom_filters(starrocks::Chunk*)
    @         0x14ee48a9 starrocks::pipeline::AggregateDistinctBlockingSourceOperator::pull_chunk(starrocks::RuntimeState*)
    @         0x15225c30 starrocks::query_cache::ConjugateOperator::pull_chunk(starrocks::RuntimeState*)
    @         0x151ee2b7 starrocks::query_cache::MultilaneOperator::_pull_chunk_from_lane(starrocks::RuntimeState*, starrocks::query_cache::MultilaneOperator::Lane&, bool)
    @         0x151eea54 starrocks::query_cache::MultilaneOperator::pull_chunk(starrocks::RuntimeState*)
    @         0x14dfd1c5 starrocks::pipeline::PipelineDriver::process(starrocks::RuntimeState*, int)
    @         0x14daff82 starrocks::pipeline::GlobalDriverExecutor::_worker_thread()
    @         0x14dae2fa starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}::operator()() const
    @         0x14dba4a2 void std::__invoke_impl<void, starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}&>(std::__invoke_other, starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}&)
    @         0x14db987b std::enable_if<is_invocable_r_v<void, starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}&>, void>::type std::__invoke_r<void, starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}&>(starrocks::pipeline::GlobalDriver_^A
    @         0x14db91fd std::_Function_handler<void (), starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}>::_M_invoke(std::_Any_data const&)
    @         0x11eb18d6 std::function<void ()>::operator()() const
    @         0x1ee8ec8c starrocks::FunctionRunnable::run()
    @         0x1ee8aba4 starrocks::ThreadPool::dispatch_thread()
    @         0x1eeac1fe void std::__invoke_impl<void, void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&>(std::__invoke_memfun_deref, void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&)
    @         0x1eeabc9f std::__invoke_result<void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&>::type std::__invoke<void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&>(void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&)
    @         0x1eeaab5c void std::_Bind<void (starrocks::ThreadPool::*(starrocks::ThreadPool*))()>::__call<void, , 0ul>(std::tuple<>&&, std::_Index_tuple<0ul>)

What type of PR is this:

• BugFix
• Feature
• Enhancement
• Refactor
• UT
• Doc
• Tool

Does this PR entail a change in behavior?

• Yes, this PR will result in a change in behavior.
• No, this PR will not result in a change in behavior.

If yes, please specify the type of change:

• Interface/UI changes: syntax, type conversion, expression evaluation, display information
• Parameter changes: default values, similar parameters but with different default values
• Policy changes: use new policy to replace old one, functionality automatically enabled
• Feature removed
• Miscellaneous: upgrade & downgrade compatibility, etc.

Checklist:

• I have added test cases for my bug fix or my new feature
• This pr needs user documentation (for new or modified features or behaviors)
  • I have added documentation for my new feature or new function
• This is a backport pr

Bugfix cherry-pick branch check:

• I have checked the version labels which the pr will be auto-backported to the target branch
  • 3.4
  • 3.3
  • 3.2
  • 3.1
  • 3.0

[github-actions]

[Java-Extensions Incremental Coverage Report]

✅ pass : 0 / 0 (0%)

[github-actions]

[FE Incremental Coverage Report]

✅ pass : 0 / 0 (0%)

[github-actions]

[BE Incremental Coverage Report]

✅ pass : 13 / 13 (100.00%)

file detail

	path	covered_line	new_line	coverage	not_covered_line_detail
🔵	be/src/exec/pipeline/operator.cpp	3	3	100.00%	[]
🔵	be/src/exec/query_cache/multilane_operator.cpp	2	2	100.00%	[]
🔵	be/src/exec/query_cache/conjugate_operator.cpp	2	2	100.00%	[]
🔵	be/src/exec/pipeline/operator.h	2	2	100.00%	[]
🔵	be/src/exec/pipeline/bucket_process_operator.cpp	4	4	100.00%	[]

[github-actions]

@Mergifyio backport branch-3.4

[github-actions]

@Mergifyio backport branch-3.3

[mergify]

backport branch-3.4

✅ Backports have been created

• #54378 [BugFix] Fix query cache crash when enable group execution (backport #54363) has been created for branch branch-3.4

[mergify]

backport branch-3.3

✅ Backports have been created

• #54379 [BugFix] Fix query cache crash when enable group execution (backport #54363) has been created for branch branch-3.3