Make aggregate state merge parallel for key8 and key16.

[incfly]

Signed-off-by: Jianfei Hu hujianfei258@gmail.com

Resolves #63666 Implemented the idea from #63667 (comment)

• Adds a method mergeToViaIndexFilter to FixedHashMap
• Create multiple worker during pipeline, 1:1 to the max_threads. Each thread uses their own thread id to invoke mergeToViaIndexFilter to avoid race condition on the fixed hash map.

We can see noticeable performance improvement for the original query, which uses fixed hash map during aggregation.

Changelog category (leave one):

• Performance Improvement

Changelog entry

When a query uses fixed hash map for aggregation state(group by a small integer), ClickHouse would merge the aggregation state in parallel to speed up the query. Resolves #63666

[clickhouse-gh]

Workflow [PR], commit [8b9cac5]

Summary: ❌

job_name	test_name	status	info	comment
Stateless tests (amd_ubsan, parallel)		failure
	00900_long_parquet_load	FAIL	cidb, flaky
Integration tests (amd_asan, old analyzer, 4/6)		failure
	test_scheduler_cpu_preemptive/test.py::test_downscaling[cpu-slot-preemption-timeout-1ms]	FAIL	cidb
	test_backup_restore_on_cluster/test_slow_rmt.py::test_replicated_database_async	FAIL	cidb
	OOM in dmesg	FAIL	cidb
Stress test (arm_asan, s3)		failure
	Server died	FAIL	cidb
	Hung check failed, possible deadlock found (see hung_check.log)	FAIL	cidb
	Killed by signal (in clickhouse-server.log)	FAIL	cidb
	Fatal message in clickhouse-server.log (see fatal_messages.txt)	FAIL	cidb
	Killed by signal (output files)	FAIL	cidb

[nickitat]

pls check sanitizer errors:
https://s3.amazonaws.com/clickhouse-test-reports/json.html?PR=87366&sha=7cc9b44ea4a1eb1997e8ebfab3bed36554e5fec7&name_0=PR&name_1=Stateless+tests+%28amd_tsan%2C+parallel%2C+1%2F2%29

[incfly]

Yeah. That's because we check fixed hashmap size during the merge, via checkLimits which require iterate all element in the buffer to see if they are zero. Meanwhile other threads are adding new entries to the first map.

I think we might be able to skip checkLimit()/size() at all, though this means more merges are executed potentially. Need to check how two level merge handles this.

[nickitat]

Yeah. That's because we check fixed hashmap size during the merge, via checkLimits which require iterate all element in the buffer to see if they are zero. Meanwhile other threads are adding new entries to the first map.

I think we might be able to skip checkLimit()/size() at all, though this means more merges are executed potentially. Need to check how two level merge handles this.

It's fine to do the check on a single thread once a merge is done

[incfly]

03373_named_session_try_recreate_before_timeout failed flaky tracking issue #87869

[incfly]

Hi @nickitat this is ready for another look, thanks!

[incfly]

test_ddl_worker_with_loopback_hosts is related to #88404

[incfly]

Hi @nickitat all comments have been addressed, PTAL, thanks!

[nickitat]

great work!