Evaluate PromQL topk/bottomk/limitk with a streaming O(T * k) plan

[nikitamikhaylov]

Changelog category (leave one):

• Performance Improvement

Changelog entry (a user-readable short description of the changes that goes into CHANGELOG.md):

Evaluate the PromQL aggregation operators topk, bottomk and limitk with a streaming plan whose selection state is O(time_steps * k) instead of collecting all series into a single row with O(time_steps * series^2) intermediate memory. Queries over high-cardinality metrics that previously failed with MEMORY_LIMIT_EXCEEDED now run in bounded memory, and the evaluation is parallelized instead of single-threaded.

Description

Closes #112335.

The transpiler evaluated topk/bottomk/limitk by collapsing the whole per-series grid into a single row (arrayTranspose(groupArray(values))) and selecting per step with arrayMap(v -> arraySort(arrayTopK(i -> v[i], k, arrayEnumerate(v))), values). The lambda captures v (an N-element array) and captured columns are replicated per array element, so one time step materializes an N x N matrix — O(T * N^2) memory independent of k, all in one row processed by one thread. On a TimeSeries table with 62.5 billion samples, topk(10, sum by(ns, pod)(rate(...))) over 25,600 series attempted a 147.71 GiB allocation for a 30-minute range and 3.43 TiB for a 30-day range.

New experimental aggregate functions (gated like the rest of the timeSeries* family) select in streaming fashion with one bounded heap of (rank, group) per time step — O(T * k) state, ~450 KiB for T = 2,880, k = 10 — and deterministic tie-breaking (smaller group wins; the old plan's tie-break was arrival-order-dependent):

• timeSeriesSelectTopKGroups(group, values, k) / timeSeriesSelectBottomKGroups(...) / timeSeriesSelectLimitKGroups(..., sampling_key)

k may be a literal, a scalar subquery, or an Array(UInt64) with one value per step (time-dependent PromQL k). The result is per selected series a per-step 0/1 mask, and the plan becomes:

-- select up to k series per step (one row per aggregation group)
SELECT timeSeriesSelectTopKGroups(group, values, <k>) AS selected_groups FROM <grid> [GROUP BY <by_expr>]
-- unfold, then mask: the mask is an arrayMap argument, never lambda-captured, so nothing replicates
SELECT group, arrayMap((x, m) -> if(m, x, NULL), values, steps_mask) AS values
FROM <grid> ANY INNER JOIN <selected> ON group = join_group

The grid is evaluated twice (selection + masking pass) — a deliberate recompute-for-memory trade; every stage streams in parallel. NULL cells are never candidates and NaN loses to any non-NaN value, matching arrayTopK/arrayBottomK and Prometheus. Exactly k series per step, NULLs where a series is not selected — byte-compatible with the old masking semantics. Serialized states carry a format version and validate wire data (size caps, per-step count <= k).

Measured on the 62.5-billion-sample table (all other query fingerprints unchanged):

shape (25,600 series)	master	this PR
topk(10, sum by(ns,pod)(rate(...))), 30 m range	OOM: 147.71 GiB attempt	5.2 s, 543.8 MiB peak
same, 30 d range	OOM: 3.43 TiB attempt	80.3 s, 44.3 GiB peak

The 30-day peak is the underlying rate grid aggregation state (identical to a plain 30-day sum(rate(...)) on this table); the selection state itself is ~450 KiB.

Verification: 17 end-to-end topk/bottomk/limitk queries (range + instant, by/without, k > N, k = 0, negative k, per-step k, scalar-subquery k, NaN, no-match metric) are byte-identical to the base binary; results are deterministic across max_threads 1/4/16 including a tie-heavy stress; all PromQL stateless tests pass with both analyzers.

Tests: a new stateless test covering the aggregate functions directly (ties, NaN, NULL gaps, serialization round-trip, corrupted-state rejection, argument validation) and the operators end to end, including a memory regression — topk(10, ...) over 3,000 series x 200 steps under max_memory_usage = 2 GB, where the old plan attempts a single 13.41 GiB allocation and the new plan peaks at 91.5 MiB — plus tests/performance/promql_topk.xml (4,000 series, 300-step ranges).

🤖 Generated with Claude Code

[clickhouse-gh]

Workflow [PR], commit [3231ae1]

Summary: ❌

• Performance Comparison: Performance dashboard

job_name	test_name	status	info	comment
Stateless tests (amd_debug, flaky check)		FAIL
	04811_promql_topk_bottomk_limitk_streaming	FAIL	cidb
	04811_promql_topk_bottomk_limitk_streaming	FAIL	cidb
Stress test (arm_msan)		FAIL
	Logical error: Shard number is greater than shard count: shard_num=A shard_count=B cluster=C (STID: 5066-2741)	FAIL	cidb
Performance Comparison (amd_release, master_head, 2/6)		FAIL	Performance dashboard
	Check Results	FAIL
Performance Comparison (arm_release, master_head, 2/6)		FAIL	Performance dashboard
	insert_sequential_and_background_merges #0::old	FAIL	query history
	insert_sequential_and_background_merges #0::new	FAIL	query history
	re2_regex_caching #3::old	FAIL	query history
	re2_regex_caching #3::new	FAIL	query history

---

AI Review

Summary

This PR replaces the old PromQL topk/bottomk/limitk plan with new timeSeriesSelect*Groups aggregates and a join-back flow, which removes the previous O(T * N^2) argument-materialization blowup. The remaining issue is that the aggregate finalization step still materializes one full steps_mask per selected series, so the new plan can reintroduce a large single-row intermediate on churny workloads and does not actually preserve the advertised bounded-memory O(T * k) behavior end to end.

Findings

⚠️ Majors

• [src/AggregateFunctions/TimeSeries/AggregateFunctionTimeSeriesSelectGroups.h:291] insertResultInto allocates a num_steps-wide UInt8 mask for every distinct selected group. When the winners change across time steps, the selected_groups intermediate grows to O(num_steps * distinct_selected_groups) rather than O(num_steps * k), so long-range high-cardinality queries can still hit MEMORY_LIMIT_EXCEEDED before the arrayJoin/join-back phase. With the PR's 30-day / 1-second example and 25,600 distinct winners over the range, this line alone would materialize about 61.8 GiB of masks. Suggested fix: keep the selection output in a step-oriented representation through the join-back, or otherwise avoid building one full-length mask per kept series.

Tests

• ⚠️ Add a focused memory regression where the winning series change across time steps, not just a monotonic value = id dataset. The current stateless memory test keeps the same topk winners for every step, so it does not exercise the selected_groups materialization path that can grow with the number of distinct winners.

Final Verdict

Changes requested.

LLVM Coverage Report

Metric	Baseline	Current	Δ
Lines	86.50%	86.50%	+0.00%
Functions	91.90%	91.80%	-0.10%
Branches	78.80%	78.80%	+0.00%

Changed lines: Changed C/C++ lines covered: 376/390 (96.41%) · Uncovered code

Full report · Diff report

[clickhouse-gh]

📊 Cloud Performance Report

✅ AI verdict: no_change — no significant changes across 36 queries analysed

This change is entirely scoped to experimental PromQL topk/bottomk/limitk streaming evaluation: new timeSeriesSelect*Groups aggregate functions (gated behind experimental settings) plus the TimeSeries query-plan builder. None of the flagged ClickBench (Q17, Q28, Q32) or TPC-H (Q4, Q7) queries execute that code path, so none of the deltas can plausibly come from this PR. Q17's flagged +11.9% is therefore downgraded to not_sure as run-to-run variance, and the TPC-H Q4/Q7 envelope suppressions are consistent with the data since both source medians fall inside master's current variance band. No cross-query operator pattern is present; these are independent off-path fluctuations.

clickbench

⚠️ 3 inconclusive

Flagged queries (3 of 43)

	Query	Verdict	Baseline median (ms)	PR median (ms)	Change	q-value	Hint
⚠️	17	not_sure	468	524	+11.9%	<0.0001	This PR only adds experimental PromQL topk/bottomk/limitk streaming aggregates and TimeSeries query-plan wiring; ClickBench Q17 executes none of that path, so the +11.9% is run-to-run variance, not a PR effect.
⚠️	28	not_sure	1036	1094	+5.6%	<0.0001	Off-path for a PromQL/TimeSeries-only change, and the two tests disagree; the +5.6% reads as measurement variance rather than a real shift.
⚠️	32	not_sure	1308	1385	+5.9%	0.0011	Nothing this PR changes runs in Q32's path, and the tests disagree; treat the +5.9% as noise.

_{Change = percent below ×2; the ratio of medians (×N faster/slower) beyond, where percent understates the scale. q-value = BH-FDR adjusted p; smaller is stronger evidence. MIRAI flags a query when q < fdr_q (default 0.10) — the value the verdict is based on.}

tpch_adapted_1_official

⚠️ 2 inconclusive

Flagged queries (2 of 22)

	Query	Verdict	Baseline median (ms)	PR median (ms)	Change	q-value	Hint
⚠️	4	not_sure	1073	1131	+5.4%	<0.0001	Envelope suppression is consistent: source median 1131 ms sits inside master's current variance band [798, 1177.5]. This PR touches only PromQL/TimeSeries code Q4 never executes.
⚠️	7	not_sure	557	627	+12.6%	<0.0001	Envelope suppression is consistent: source median 627 ms is inside master's band [552, 667]. Q7 does not run the PromQL/TimeSeries path this PR changes, so the +12.6% is not attributable to the diff.

_{Change = percent below ×2; the ratio of medians (×N faster/slower) beyond, where percent understates the scale. q-value = BH-FDR adjusted p; smaller is stronger evidence. MIRAI flags a query when q < fdr_q (default 0.10) — the value the verdict is based on.}

Debug info

• StressHouse run: 60863034-56d4-4007-8b46-5524ead0f758
• MIRAI run: 9284d40b-20e3-4892-855d-d6e09babda94
• PR check IDs:
  • clickbench_176149_1786110839
  • clickbench_176160_1786110839
  • clickbench_176172_1786110839
  • tpch_adapted_1_official_176179_1786110839
  • tpch_adapted_1_official_176205_1786110839
  • tpch_adapted_1_official_176219_1786110839

[nikitamikhaylov]

@vitlibar I tried to understand whether we can change the original arrayTopK/arrayBottomK functions, but looks like no:

arrayTopK itself was never the memory problem — internally it already does a bounded O(k) selection over its input array. The O(T * N^2) blowup in the old plan happened before arrayTopK ever ran, in how its arguments were materialized:

1. Lambda capture replication. The old plan computed arrayMap(v -> arraySort(arrayTopK(i -> v[i], k, arrayEnumerate(v))), values). In ClickHouse, a column captured by a higher-order-function lambda is physically replicated once per element of the array being mapped over. Here the captured v is an N-element array (one value per series) and the map runs over N indices per time step — so the engine materializes an N×N matrix per step as the function's input. A heap inside arrayTopK can't help; the 147 GiB allocation happens while building its argument columns, before its algorithm executes a single instruction.
2. Single-row collapse. Even if the capture problem were engineered away (and the capture is needed there — the lambda ranks indices by value to get which series won, not just the winning values), the old plan still required arrayTranspose(groupArray(...)): all N series collected into one row, processed by one thread. That's O(T * N) in a single value with no streaming and no parallelism — still gigabytes at high cardinality, and single-threaded.

[clickhouse-gh]

insertResultInto re-expands the bounded heap state into one full steps_mask per distinct selected series, so the selected_groups cell becomes O(num_steps * distinct_selected_groups), not O(num_steps * k). In the churny case where different series win at different time steps, distinct_selected_groups grows to min(num_series, num_steps * k), and this line allocates a UInt8 mask of length num_steps for each of them.

That means the rewritten PromQL plan can still materialize a very large single-row intermediate before the arrayJoin/join-back phase. With the PR's own 30-day / 1-second example, num_steps = 2_592_000; if the winners churn enough to touch all 25,600 series over the range, this path alone builds about 61.8 GiB of masks here. The existing memory regression test keeps the same winners at every step (value = id), so it does not exercise this case.

I think the selection result needs to stay step-oriented through the join-back, or at least this needs a churny memory regression test before we claim the new plan is bounded by O(T * k).