Aggregator support prefetch

[guo-shaoge]

What problem does this PR solve?

Issue Number: close #9680

Problem Summary:

What is changed and how it works?

1. Support prefetch for HashTable. This PR mainly works for key32/key64/key128/key256, and it doesn't affect key_string/key_serialized

Benchmark

Workloads: TPCH-50G
Queries:

 # --q0 Q1-1: key_int64; distinct rate: 10M/300M; HashMap
  "select /*+ mpp_1phase_agg() */ sum(l_quantity) as csum, l_partkey from lineitem group by l_partkey having csum < 200;"
  # --q1 Q1-2: key_int64; 75M/300M; HashMp
  "select /*+ mpp_1phase_agg() */ sum(l_quantity) as csum, l_orderkey from lineitem group by  L_orderkey having csum < 1;"
  # --q2 Q1-3: key_int64; 7/300M; HashMap
  "select /*+ mpp_1phase_agg() */ sum(l_quantity), l_linenumber from lineitem group by  l_linenumber;"

  # --q3 Q2-1: one_key_strbinpadding_phmap; 2/300M; StringHashMap
  "select /*+ mpp_1phase_agg() */ sum(l_discount), l_linestatus from lineitem group by l_linestatus;"
  # --q4 Q2-2: one_key_strbinpadding; 104M/300M; StringHashMap
  "select /*+ mpp_1phase_agg() */ sum(l_discount) as csum from lineitem group by l_comment having csum > 800;"

  # --q4 Q3-1: key_serialized as group by method;  33/300M; HashMap with StringRef key
  "select /*+ mpp_1phase_agg() */ sum(l_discount), l_returnflag from lineitem group by l_returnflag, l_discount;"
  # --q6 Q3-2: key_serialized as group by method; 77M/300M; HashMap with StringRef key
  "select /*+ mpp_1phase_agg() */ sum(l_discount) as csum, l_returnflag from lineitem group by l_returnflag, l_discount, l_extendedprice having csum > 100;"


  # --q7 Q4-1: two_keys_num64_strbinpadding: 21/300M; HashMap with StringRef key
  "select /*+ mpp_1phase_agg() */ sum(l_discount) from lineitem group by l_returnflag, L_LINENUMBER;"
  # --q8 Q4-2: two_keys_num64_strbinpadding; 29.9M/300M; HashMap with StringRef key
  "select /*+ mpp_1phase_agg() */ sum(l_discount) as csum, l_partkey from lineitem group by l_returnflag, l_partkey having csum > 100;"

  # --q9 Q5-1: keys_128; 77/300M; HashMap with UInt128 key
  "select /*+ mpp_1phase_agg() */ sum(l_linenumber), l_discount from lineitem group by l_linenumber, l_discount;"
  # --q10 Q5-2: keys_128; 5M/300M; HashMap with UInt128 key
  "select /*+ mpp_1phase_agg() */ sum(l_linenumber) as csum, l_discount from lineitem group by l_suppkey, l_discount having csum > 800;"

  # --q11 Q6-1: key_string; 4/300M; StringHashMap
   "select /*+ mpp_1phase_agg() */ sum(l_quantity), l_shipinstruct from lineitem group by l_shipinstruct;"

  # --q12 Q7-1: keys_256; 290M/300M; HashMap;
  "select /*+ mpp_1phase_agg() */ sum(l_suppkey) as csum from lineitem group by l_suppkey, l_tax, l_discount, l_partkey having csum = 10;"

  # --q13 clickbench Q4
  "select count(distinct userid) from clickbench.hits;"

ENV:

1. Internal benchbot env
2. 1 tidb; 2 tikv; 1 PD; 2 tiflash(16C64G, amd64)

Results (google excel):

idx	query name	nightly-b16a5f93	this PR - opt1	this PR - opt2	rate-1	rate-2
0	Q1-1	2.71	2.05	1.8	24.35%	33.58%
1	Q1-2	2.15	2.07	1.62	3.72%	24.65%
2	Q1-3	1.33	1.4	1.31	-5.26%	1.50%
9	Q5-1	1.37	1.35	1.37	1.46%	0.00%
10	Q5-2	4.08	3.86	3.59	5.39%	12.01%
12	Q7-1	13.3	12.1	11.2	9.02%	15.79%
13	clickbench Q4	0.436	0.383	0.391	12.16%	10.32%

NOTE:

1. Query duration is in seconds
2. Queries like Q2-x/Q3-x are not listed because they are key_stirng/key_serialized, this PR only works for key_int64/key_int128/key_int256...). So these queries should not be affected.
3. opt-1 means only prefetch HashTable.
4. opt-2 means(this is the final improvement of this PR)
   1. prefetch HashTable
   2. prefetch threshold be 2MB in byte instead of 8192 buckets
   3. prefetch when insertResultInto columns
   4. prefetch aggregation data for already exists key in HashTable
   5. mini batch(step is 256 for now) when emplace HashTable
5. ClickBench Q4 is from: SELECT COUNT(DISTINCT UserID) FROM hits; https://github.com/ClickHouse/ClickBench/blob/main/mysql/queries.sql

Check List

Tests

• Unit test
• Integration test
• Manual test (add detailed scripts or steps below)
• No code

Side effects

• Performance regression: Consumes more CPU
• Performance regression: Consumes more Memory
• Breaking backward compatibility

Documentation

• Affects user behaviors
• Contains syntax changes
• Contains variable changes
• Contains experimental features
• Changes MySQL compatibility

Release note

None

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please ask for approval from guo-shaoge, ensuring that each of them provides their approval before proceeding. For more information see the Code Review Process.

The full list of commands accepted by this bot can be found here.

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[guo-shaoge]

/retest

[windtalker]

I think just use auto last_offset = offsets[row - 1] is ok

[guo-shaoge]

/retest

[gengliqi]

Suggested change

	void ALWAYS_INLINE prefetch(size_t hashval) const
	void ALWAYS_INLINE prefetchRead(size_t hashval) const

[guo-shaoge]

I think prefetch is simple and clear

[guo-shaoge]

todo: why force_agg_prefetch in ComputeServerRunner.testErrorMessage will hang

[guo-shaoge]

from @windtalker : setCancelTest() will make table scan output infinite output data, so this case will run forever if this case doesn't report error

[guo-shaoge]

Suggested change

	ALWAYS_INLINE inline size_t getCurrentHashAndSetUpPrefetchHash(
	ALWAYS_INLINE inline size_t getCurrentRowHashAndDoPrefetch(