ClickHouse distributed GROUP BY partial results under timeout_overflow_mode break
0 reputation · 04 May 2020, 19:38 UTC
0 reputation · 04 May 2020, 19:38 UTC
I need to return partial aggregation results from a ClickHouse distributed GROUP BY query when execution exceeds max_execution_time, using timeout_overflow_mode='break' so the coordinator stops the query instead of throwing.
Cancellation is propagated to remote shards asynchronously, and some shards may finish aggregation while others are interrupted mid-aggregation. ClickHouse does not document a deterministic merge rule for these partial aggregates, so result consistency across repeated runs is uncertain.
GROUP BY results when timeout_overflow_mode='break' interrupts shards unevenly?35775 reputation · 04 May 2020, 20:37 UTC
No. ClickHouse does not guarantee deterministic merging of partial GROUP BY results when timeout_overflow_mode='break' interrupts shards unevenly. In ClickHouse 23.x and later, treat this mode as a best-effort partial-result mechanism, not a correctness-stable contract. Avoid it for exact billing, compliance, or SLA aggregates.
In a distributed GROUP BY, each shard computes local partial aggregate states and sends them to the initiator. With timeout_overflow_mode='break', the initiator stops waiting and merges whichever states have arrived. Because cancellation is asynchronous, shards interrupted mid-aggregation may contribute partial states, no states, or fail to transmit. The set of contributing shards can therefore vary between runs.
count, sum, min, max) merge correctly only for states that arrive. Missing shard data yields missing groups or undercounts.ORDER BY, output row order is unspecified. Repeated timed runs can differ in row count and values.distributed_aggregation_memory_efficient. Do not assume 23.x behavior is identical across later releases without testing on your target version.If deterministic results are required, let the query complete:
max_execution_time or remove the timeout for correctness-critical queries.To confirm nondeterminism on your cluster, run the same timed query repeatedly on unchanged data and compare row counts, group checksums, and aggregate sums:
SELECT key, count(), sum(metric)
FROM distributed_table
GROUP BY key
SETTINGS max_execution_time = 5, timeout_overflow_mode = 'break';Inspect system.query_log for per-shard elapsed times and cancellation events. Compare against a baseline run with a higher timeout to quantify missing groups and undercounts.
Use comments to ask for clarification. Post a solution as an answer.
6,300 reputation · 05 May 2020, 03:08 UTC
Use the query‑profile log to confirm that the coordinator receives complete partial aggregates from each shard before the abort; if any shard contributes an incomplete state, the final result will differ across executions. To obtain deterministic output, either raise max_execution_time, switch to timeout_overflow_mode='continue', or pre‑aggregate on shards and merge locally after the query finishes.