This patch adds parallel execution for the ORDER BY + LIMIT (top K) phase in high-cardinality parallel GROUP BY queries. When the GROUP BY result is sharded (due to high cardinality), the top K selection now processes each shard in parallel using worker threads, then merges the per-shard results.

Changes:

• New GroupByLongTopKJob and GroupByLongTopKTask for parallel top K processing
• New message bus queue for distributing top K tasks to workers
• AsyncGroupByRecordCursor#parallelLongTopK() orchestrates parallel execution when:
  • The ORDER BY function is thread-safe
  • The map is sharded
  • Result size exceeds cairo.sql.parallel.groupby.topk.threshold
• New configuration properties:
  • cairo.sql.parallel.groupby.topk.threshold - minimum map size to enable parallel top K (default 5M)
  • cairo.sql.parallel.groupby.topk.queue.capacity - task queue capacity (default - same as page frame reduce queue capacity)

Benchmarks

ClickBench run on Ryzen 7900x, 64GB RAM, Ubuntu 24.04.

The queries that benefit from parallel top K and their respective sharded map sizes:

• Q12 - 6,019,102
• Q14 - 6,474,212
• Q15 - 17,630,976
• Q16 - 24,070,560
• Q18 - 56,384,822
• Q30 - 5,730,331
• Q31 - 13,172,392
• Q32 - 99,997,493
• Q33 - 18,342,019
• Q34 - 18,342,019
• Q35 - 9,762,046

ORDER BY + LIMIT phase (not full query execution) times:

Analysis of Q32 speed-up

Memory bandwidth analysis:

For Q32's map entries (GROUP BY WatchID long, ClientIP ipv4):

• Key: ~12 bytes (long + ipv4)
• Values: COUNT(*) + SUM + AVG internals ≈ 32+ bytes
• ~48 bytes per entry (with alignment)

Total data: 100M × 48 bytes ≈ 4.8 GB

Ryzen 7900X with DDR5 dual-channel has theoretical bandwidth of ~90 GB/s, practical peak ~60-70 GB/s. The sequential scan is already at ~60-70% of peak memory bandwidth. The parallel version pushes it closer to ~75-80%. We're hitting the memory bandwidth ceiling.

This explains the modest 1.3x improvement - it's not a CPU-bound workload where 23 workers would help proportionally. The single-threaded scan with hardware prefetching already saturates most of the available DRAM bandwidth. Additional threads only marginally improve memory-level parallelism. The queries with better speedups (Q35 at 11.2x) have smaller working sets that benefit more from cache effects and CPU parallelism rather than being memory-bound.

Queries

-- Q12
SELECT SearchPhrase, COUNT(*) AS c FROM hits WHERE SearchPhrase IS NOT NULL GROUP BY SearchPhrase ORDER BY c DESC LIMIT 10;

-- Q14
SELECT SearchEngineID, SearchPhrase, COUNT(*) AS c FROM hits WHERE SearchPhrase IS NOT NULL GROUP BY SearchEngineID, SearchPhrase ORDER BY c DESC LIMIT 10;

-- Q15
SELECT UserID, COUNT(*) AS c FROM hits GROUP BY UserID ORDER BY c DESC LIMIT 10;

-- Q16
SELECT UserID, SearchPhrase, COUNT(*) AS c FROM hits GROUP BY UserID, SearchPhrase ORDER BY c DESC LIMIT 10;

-- Q18
SELECT UserID, extract(minute FROM EventTime) AS m, SearchPhrase, COUNT(*) AS c FROM hits GROUP BY UserID, m, SearchPhrase ORDER BY c DESC LIMIT 10;

-- Q30
SELECT SearchEngineID, ClientIP, COUNT(*) AS c, SUM(IsRefresh), AVG(ResolutionWidth) FROM hits WHERE SearchPhrase IS NOT NULL GROUP BY SearchEngineID, ClientIP ORDER BY c DESC LIMIT 10;

-- Q31
SELECT WatchID, ClientIP, COUNT(*) AS c, SUM(IsRefresh), AVG(ResolutionWidth) FROM hits WHERE SearchPhrase IS NOT NULL GROUP BY WatchID, ClientIP ORDER BY c DESC LIMIT 10;

-- Q32
SELECT WatchID, ClientIP, COUNT(*) AS c, SUM(IsRefresh), AVG(ResolutionWidth) FROM hits GROUP BY WatchID, ClientIP ORDER BY c DESC LIMIT 10;

-- Q33
SELECT URL, COUNT(*) AS c FROM hits ORDER BY c DESC LIMIT 10;

-- Q35
SELECT ClientIP, ClientIP - 1, ClientIP - 2, ClientIP - 3, COUNT(*) AS c FROM hits GROUP BY ClientIP, ClientIP - 1, ClientIP - 2, ClientIP - 3 ORDER BY c DESC LIMIT 10;