[BugFix] Fix sequence of selection and input_chunk for streaming_agg_with_selection (backport #54595)

[mergify]

Why I'm doing:

Previous Logic:

1. evaluate_group_by_columns.
2. evaluate_input_columns
3. calculate _streaming_selection to indicate which rows hit HT.
4. output_chunk_by_streaming_with_selection (filter group_by_columns and input_columns by _streaming_selection)
5. output_chunk_by_streaming

Recent Logic:

1. evaluate_group_by_columns
2. calculate _streaming_selection to indicate which rows hit HT.
3. output_chunk_by_streaming_with_selection (filter group_by_columns by _streaming_selection)
4. output_chunk_by_streaming (evaluate input_columns here now)

This cause two issues.

1. In output_chunk_by_streaming, before calling evaluate_input_columns, the number of rows in group_by_columns and input_chunk differs. This is because group_by_columns is filtered by _streaming_selection, whereas input_chunk is not.

2. The result of evaluate_input_columns is not filtered by _streaming_selection.

*** Aborted at 1735817590 (unix time) try "date -d @1735817590" if you are using GNU date ***
PC: @     0x7ff6544c3387 __GI_raise
*** SIGABRT (@0x3e80000448c) received by PID 17548 (TID 0x7ff56c586700) from PID 17548; stack trace: ***
    @         0x1c6fcf47 google::(anonymous namespace)::HandleSignal(int, siginfo_t*, void*)
    @     0x7ff65695920b __pthread_once_slow
    @         0x1c6fc754 google::(anonymous namespace)::FailureSignalHandler(int, siginfo_t*, void*)
F20250102 19:33:10.237299 140692041484032 aggregator.cpp:951] Check failed: num_rows == _group_by_columns[i]->size() (3014 vs. 2357)
    @     0x7ff656962630 (/usr/lib64/libpthread-2.17.so+0xf62f)
    @     0x7ff6544c3387 __GI_raise
    @     0x7ff6544c4a78 __GI_abort
    @          0xd76fc0a starrocks::failure_function()
    @         0x1c6f154a google::LogMessage::Fail()
    @         0x1c6f2f84 google::LogMessageFatal::~LogMessageFatal()
    @          0xf5700ac starrocks::Aggregator::output_chunk_by_streaming(starrocks::Chunk*, std::shared_ptr<starrocks::Chunk>*)
    @          0xf5730bc starrocks::Aggregator::output_chunk_by_streaming_with_selection(starrocks::Chunk*, std::shared_ptr<starrocks::Chunk>*)
    @          0xf1612ae starrocks::pipeline::AggregateStreamingSinkOperator::_push_chunk_by_selective_preaggregation(std::shared_ptr<starrocks::Chunk> const&, unsigned long, bool)
    @          0xf163d30 starrocks::pipeline::AggregateStreamingSinkOperator::_push_chunk_by_auto(std::shared_ptr<starrocks::Chunk> const&, unsigned long)
    @          0xf15eba3 starrocks::pipeline::AggregateStreamingSinkOperator::push_chunk(starrocks::RuntimeState*, std::shared_ptr<starrocks::Chunk> const&)
    @          0xf0c306c starrocks::pipeline::PipelineDriver::process(starrocks::RuntimeState*, int)
    @          0xfed09a3 starrocks::pipeline::GlobalDriverExecutor::_worker_thread()
    @          0xfecf1c6 starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}::operator()() const
    @          0xfed99ec void std::__invoke_impl<void, starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}&>(std::__invoke_other, starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}&)
    @          0xfed8dca 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@
    @          0xfed877f std::_Function_handler<void (), starrocks::pipeline::GlobalDriverExecutor::initialize(int)::{lambda()#1}>::_M_invoke(std::_Any_data const&)
    @          0xbdcd888 std::function<void ()>::operator()() const
    @          0xc595650 starrocks::FunctionRunnable::run()
    @          0xc592494 starrocks::ThreadPool::dispatch_thread()
    @          0xc5b1ae0 void std::__invoke_impl<void, void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&>(std::__invoke_memfun_deref, void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&)
    @          0xc5b03f3 std::__invoke_result<void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&>::type std::__invoke<void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&>(void (starrocks::ThreadPool::*&)(), starrocks::ThreadPool*&)
    @          0xc5aee80 void std::_Bind<void (starrocks::ThreadPool::*(starrocks::ThreadPool*))()>::__call<void, , 0ul>(std::tuple<>&&, std::_Index_tuple<0ul>)
    @          0xc5ad442 void std::_Bind<void (starrocks::ThreadPool::*(starrocks::ThreadPool*))()>::operator()<, void>()
    @          0xc5ab4f4 void std::__invoke_impl<void, std::_Bind<void (starrocks::ThreadPool::*(starrocks::ThreadPool*))()>&>(std::__invoke_other, std::_Bind<void (starrocks::ThreadPool::*(starrocks::ThreadPool*))()>&)
    @          0xc5a8f02 std::enable_if<is_invocable_r_v<void, std::_Bind<void (starrocks::ThreadPool::*(starrocks::ThreadPool*))()>&>, void>::type std::__invoke_r<void, std::_Bind<void (starrocks::ThreadPool::*(starrocks::ThreadPool*))()>&>(std::_Bind<void (starrocks::ThreadPool:@
    @          0xc5a477f std::_Function_handler<void (), std::_Bind<void (starrocks::ThreadPool::*(starrocks::ThreadPool*))()> >::_M_invoke(std::_Any_data const&)
    @          0xbdcd888 std::function<void ()>::operator()() const
    @          0xc5785c9 starrocks::Thread::supervise_thread(void*)

What I'm doing:

Fixes https://github.com/StarRocks/StarRocksTest/issues/9009

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

[mergify]

Cherry-pick of efe1dde has failed:

On branch mergify/bp/branch-3.1/pr-54595
Your branch is up to date with 'origin/branch-3.1'.

You are currently cherry-picking commit efe1ddecb4.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Changes to be committed:
	modified:   be/src/exec/aggregator.cpp

Unmerged paths:
  (use "git add <file>..." to mark resolution)
	both modified:   be/src/exec/aggregator.h

To fix up this pull request, you can check it out locally. See documentation: https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally

[mergify]

@mergify[bot]: Backport conflict, please reslove the conflict and resubmit the pr