[Enhancement] Introduce pipeline-level multi-partitioned runtime filter(backport #33002)

[satanson]

cherry-pick #33002
Pipeline-level runtime feature supports multi-partitioned pipeline-level rf, in the old-fashioned runtime filter construction, each driver in an instance just collects runtime filter raw materials and the driver who finishes HT constuction last constructs the singleton runtime filter finally, it is quite time-consuming when the right table of HashJoin is very large. In this feature, we eliminate this issue, each driver construct its own runtime filter, the last driver only merge the all of these runtime filters into a multi-partitioned runtime-filter with negligible cost.

This feature is controled by the session variable as follows:

enable_pipeline_level_multi_partitioned_rf: true for turning on, false(in default) for turning off.

We use a flatten array to hold multi-partitioned runtime filter for both global and instance rf, while runtime filter's layout is depends on distribution mode of HashJoin, pipeline_dop and local exchange interpolation, so its layout is not very clear, RuntimeFilterLayout infered from these information is introduced to describle the layout of runtime-filter.

RuntimeFilterLayout can describe both old-fashioned runtime filter and new-fashioned pipeline-level multi-partitioned runtime filter. its options as follows:

1. SINGTETON: an partitioned runtime filter;
2. GLOBAL_SHUFFLE_1L: old-fashed grf generated by shuffle join;
3. GLOBAL_BUCKET_1L: old-fashed grf generated by bucket_shuffle/colocate join;
4. PIPELINE_SHUFFLE: pipeline-level multi-partitioned rf generated by shuffle join;
5. GLOBAL_SHUFFLE_2L: global rf constructed by concatenating PIPELINE_SHUFFLE rf(s) of each instance;
6. PIPELINE_BUCKET: pipeline-level multi-partitioned rf generated by bucket_shuffle/colocate join without local exchange interpolation;
7. PIPELINE_BUCKET_LX: pipeline-level multi-partitioned rf generated by bucket_shuffle/colocate join with local exchange interpolation;
8. GLOBAL_BUCKET_2L: global rf constructed by PIPELINE_BUCKET rf(s) of each instance;
9. GLOBAL_BUCKET_2L_LX: global rf constructed by PIPELINE_BUCKET_LX rf(s) of each instance;

For old-fashioned runtime filter, we use options as following table shows

+=================================+=============+===================+
| Join Distribution               | instance rf | globa rf          |
+=================================+=============+===================+
| BROADCAST/REPLICATED            | SINGLETON   | SINGLETON         |
| PARTITIONED/SHUFFLE_HASH_BUCKET | SINGLETON   | GLOBAL_SHUFFLE_1L |
| COLOATE/LOCAL_SHUFFLE_BUCKET    | SINGLETON   | GLOBAL_BUCKET_1L  |
+---------------------------------+-------------+-------------------+

For the multi-partitioned pipeline-level rf, we use options as following table shows

+==========================================+====================+=====================+
| Join Distribution                        | instance rf        | global rf           |
+==========================================+====================+=====================+
| BROADCAST/REPLICATED                     | SINGLETON          | SINGLETON           |
| PARTITIONED/SHUFFLE_HASH_BUCKET          | PIPELINE_SHUFFLE   | GLOBAL_SHUFFLE_2L   |
| COLOATE/LOCAL_SHUFFLE_BUCKET(without LX) | PIPELINE_BUCKET    | GLOBAL_BUCKET_2L    |
| COLOATE/LOCAL_SHUFFLE_BUCKET(with LX)    | PIPELINE_BUCKET_LX | GLOBAL_BUCKET_2L_LX |
+------------------------------------------+--------------------+---------------------+

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

Bugfix cherry-pick branch check:

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

[sonarqubecloud]

Kudos, SonarCloud Quality Gate passed!   

0 Bugs
0 Vulnerabilities
0 Security Hotspots
25 Code Smells

0.0% Coverage
0.4% Duplication

The version of Java (11.0.21) you have used to run this analysis is deprecated and we will stop accepting it soon. Please update to at least Java 17.
Read more here

[github-actions]

[FE Incremental Coverage Report]

❌ fail : 107 / 180 (59.44%)

file detail

	path	covered_line	new_line	coverage	not_covered_line_detail
🔵	com/starrocks/planner/RuntimeFilterLayout.java	0	62	00.00%	[25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 37, 41, 45, 49, 53, 57, 61, 65, 69, 73, 81, 84, 85, 86, 90, 91, 92, 93, 95, 96, 98, 102, 105, 110, 111, 112, 113, 115, 116, 118, 122, 125, 126, 127, 129, 130, 132, 134, 140, 141, 142, 143, 144, 145, 146, 147, 148, 150, 151, 153, 154, 156]
🔵	com/starrocks/qe/SessionVariable.java	6	10	60.00%	[1992, 1993, 1994, 1996]
🔵	com/starrocks/planner/RuntimeFilterDescription.java	61	67	91.04%	[288, 295, 303, 388, 407, 419]
🔵	com/starrocks/qe/CoordinatorPreprocessor.java	37	38	97.37%	[1328]
🔵	com/starrocks/qe/Coordinator.java	3	3	100.00%	[]

[github-actions]

[BE Incremental Coverage Report]

❌ fail : 12 / 113 (10.62%)

file detail

	path	covered_line	new_line	coverage	not_covered_line_detail
🔵	src/exec/hash_joiner.cpp	0	2	00.00%	[578, 580]
🔵	src/exec/pipeline/runtime_filter_types.h	0	12	00.00%	[49, 58, 316, 317, 319, 321, 322, 323, 324, 325, 326, 327]
🔵	src/exprs/runtime_filter_bank.h	0	2	00.00%	[125, 163]
🔵	src/exprs/runtime_filter_bank.cpp	0	4	00.00%	[213, 253, 466, 476]
🔵	src/runtime/runtime_filter_worker.cpp	0	2	00.00%	[324, 631]
🔵	src/exprs/runtime_filter.h	7	83	08.43%	[276, 294, 295, 329, 352, 353, 358, 359, 360, 361, 364, 365, 369, 370, 372, 373, 376, 377, 381, 384, 388, 389, 390, 393, 394, 395, 407, 415, 416, 509, 510, 511, 514, 515, 516, 517, 519, 520, 521, 522, 524, 525, 526, 529, 530, 531, 534, 537, 538, 539, 540, 541, 542, 544, 545, 546, 549, 551, 552, 553, 555, 557, 558, 559, 560, 563, 564, 565, 566, 567, 748, 749, 751, 753, 754, 756]
🔵	src/exprs/runtime_filter.cpp	5	8	62.50%	[192, 195, 211]