[Enhancement] support per bucket optimize for colocate aggregate (backport #29252)

[mergify]

This is an automatic backport of pull request #29252 done by Mergify.
Cherry-pick of 37b2b0c has failed:

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

You are currently cherry-picking commit 37b2b0c2e8.
  (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/CMakeLists.txt
	modified:   be/src/exec/aggregate/aggregate_blocking_node.cpp
	modified:   be/src/exec/aggregate/aggregate_blocking_node.h
	modified:   be/src/exec/aggregate/distinct_blocking_node.cpp
	modified:   be/src/exec/aggregate/distinct_blocking_node.h
	modified:   be/src/exec/olap_scan_node.cpp
	modified:   be/src/exec/olap_scan_node.h
	modified:   be/src/exec/pipeline/aggregate/aggregate_blocking_sink_operator.h
	modified:   be/src/exec/pipeline/aggregate/spillable_aggregate_blocking_sink_operator.cpp
	modified:   be/src/exec/pipeline/aggregate/spillable_aggregate_blocking_sink_operator.h
	new file:   be/src/exec/pipeline/bucket_process_operator.cpp
	modified:   be/src/exec/pipeline/pipeline.cpp
	modified:   be/src/exec/pipeline/pipeline_driver.cpp
	modified:   be/src/exec/pipeline/scan/chunk_source.cpp
	modified:   be/src/exec/pipeline/scan/morsel.cpp
	modified:   be/src/exec/pipeline/scan/morsel.h
	modified:   be/src/exec/pipeline/scan/scan_operator.cpp
	modified:   be/src/exec/pipeline/scan/scan_operator.h
	modified:   be/src/exec/query_cache/cache_operator.h
	modified:   be/src/exec/query_cache/multilane_operator.h
	modified:   be/src/exec/query_cache/ticket_checker.cpp
	modified:   be/src/exec/query_cache/ticket_checker.h
	modified:   be/src/exec/scan_node.cpp
	modified:   be/src/exec/scan_node.h
	modified:   be/src/exec/spill/input_stream.h
	modified:   be/src/exec/spill/spiller.cpp
	modified:   be/src/exec/spill/spiller.h
	modified:   fe/fe-core/src/main/java/com/starrocks/planner/AggregationNode.java
	modified:   fe/fe-core/src/main/java/com/starrocks/planner/OlapScanNode.java
	modified:   fe/fe-core/src/main/java/com/starrocks/planner/PlanNode.java
	modified:   fe/fe-core/src/main/java/com/starrocks/qe/CoordinatorPreprocessor.java
	modified:   fe/fe-core/src/main/java/com/starrocks/qe/SessionVariable.java
	modified:   fe/fe-core/src/main/java/com/starrocks/sql/optimizer/Optimizer.java
	modified:   fe/fe-core/src/main/java/com/starrocks/sql/optimizer/operator/physical/PhysicalHashAggregateOperator.java
	modified:   fe/fe-core/src/main/java/com/starrocks/sql/optimizer/operator/physical/PhysicalOlapScanOperator.java
	modified:   fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/tree/AddDecodeNodeForDictStringRule.java
	new file:   fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/tree/PhysicalDistributionAggOptRule.java
	deleted:    fe/fe-core/src/main/java/com/starrocks/sql/optimizer/rule/tree/UseSortAggregateRule.java

Unmerged paths:
  (use "git add/rm <file>..." as appropriate to mark resolution)
	both added:      be/src/exec/pipeline/bucket_process_operator.h
	both modified:   be/src/exec/pipeline/operator.h
	both modified:   fe/fe-core/src/main/java/com/starrocks/planner/PlanFragment.java
	deleted by us:   fe/fe-core/src/main/java/com/starrocks/qe/scheduler/assignment/LocalFragmentAssignmentStrategy.java
	both modified:   fe/fe-core/src/main/java/com/starrocks/sql/plan/PlanFragmentBuilder.java
	both modified:   gensrc/thrift/PlanNodes.thrift

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

---

Mergify commands and options

More conditions and actions can be found in the documentation.

You can also trigger Mergify actions by commenting on this pull request:

• @Mergifyio refresh will re-evaluate the rules
• @Mergifyio rebase will rebase this PR on its base branch
• @Mergifyio update will merge the base branch into this PR
• @Mergifyio backport <destination> will backport this PR on <destination> branch

Additionally, on Mergify dashboard you can:

• look at your merge queues
• generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.com

[mergify]

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

[github-actions]

[FE Incremental Coverage Report]

❌ fail : 27 / 35 (77.14%)

file detail

	path	covered_line	new_line	coverage	not_covered_line_detail
🔵	com/starrocks/sql/optimizer/rule/tree/AddDecodeNodeForDictStringRule.java	0	1	00.00%	[755]
🔵	com/starrocks/planner/OlapScanNode.java	5	11	45.45%	[198, 406, 407, 408, 468, 469]
🔵	com/starrocks/sql/optimizer/operator/physical/PhysicalOlapScanOperator.java	2	3	66.67%	[146]
🔵	com/starrocks/qe/SessionVariable.java	2	2	100.00%	[]
🔵	com/starrocks/planner/AggregationNode.java	8	8	100.00%	[]
🔵	com/starrocks/sql/plan/PlanFragmentBuilder.java	7	7	100.00%	[]
🔵	com/starrocks/sql/optimizer/operator/physical/PhysicalHashAggregateOperator.java	3	3	100.00%	[]

[github-actions]

[BE Incremental Coverage Report]

❌ fail : 41 / 155 (26.45%)

file detail

	path	covered_line	new_line	coverage	not_covered_line_detail
🔵	src/exec/aggregate/aggregate_blocking_node.cpp	0	16	00.00%	[178, 179, 213, 214, 216, 217, 224, 227, 228, 229, 250, 251, 286, 287, 292, 295]
🔵	src/exec/pipeline/aggregate/spillable_aggregate_blocking_sink_operator.h	0	1	00.00%	[61]
🔵	src/exec/pipeline/operator.h	0	1	00.00%	[262]
🔵	src/exec/spill/spiller.cpp	0	5	00.00%	[117, 119, 120, 121, 123]
🔵	src/exec/pipeline/aggregate/spillable_aggregate_blocking_sink_operator.cpp	0	6	00.00%	[113, 115, 116, 117, 119, 120]
🔵	src/exec/olap_scan_node.cpp	0	8	00.00%	[67, 68, 69, 70, 401, 402, 405, 406]
🔵	src/exec/olap_scan_node.h	0	1	00.00%	[98]
🔵	src/exec/aggregate/distinct_blocking_node.cpp	0	12	00.00%	[134, 163, 165, 167, 168, 175, 176, 178, 180, 205, 206, 228]
🔵	src/exec/pipeline/scan/morsel.cpp	8	53	15.09%	[79, 80, 81, 83, 86, 87, 90, 91, 96, 106, 148, 149, 151, 152, 155, 156, 159, 161, 162, 166, 167, 170, 171, 172, 173, 175, 176, 179, 180, 181, 182, 183, 184, 186, 187, 189, 190, 194, 197, 198, 199, 200, 202, 203, 204]
🔵	src/exec/scan_node.cpp	1	3	33.33%	[143, 144]
🔵	src/exec/pipeline/scan/morsel.h	8	19	42.11%	[139, 224, 225, 226, 304, 309, 310, 313, 314, 315, 318]
🔵	src/exec/pipeline/pipeline_driver.cpp	1	2	50.00%	[104]
🔵	src/exec/pipeline/scan/scan_operator.cpp	9	13	69.23%	[455, 456, 457, 458]
🔵	src/exec/pipeline/scan/chunk_source.cpp	3	4	75.00%	[79]
🔵	src/exec/spill/spiller.h	2	2	100.00%	[]
🔵	src/exec/pipeline/scan/scan_operator.h	1	1	100.00%	[]
🔵	src/exec/query_cache/ticket_checker.cpp	4	4	100.00%	[]
🔵	src/exec/pipeline/pipeline.cpp	4	4	100.00%	[]