Nightly uniqExact report hits the memory limit on a distributed join

The 30-day window is the trigger and the aggregation is the consumer. uniqExact(user_id) inside a GROUP BY page_url keeps an exact hash set of distinct users for every group; over 30 days those sets are large, and the aggregation state is what reaches 34 GiB. The join is not the main culprit - that is what the fast single-day run is telling you, since the join shape is identical.

Order of attack:

1. Filter earlier. Move event_time, device and bot exclusions into PREWHERE and confirm with EXPLAIN indexes = 1 that only the intended monthly partitions are read. Fewer input rows means smaller aggregation states.
2. Shrink the grouping key before the join. Group by page_id instead of the wide page_url string, then join the small result to pages at the end. The join then operates on thousands of rows instead of billions, and the hash table holds integers rather than URLs.
3. Right-size the distinct count. uniqCombined(12) costs a fraction of the memory for roughly 0.5% error. If the number must be auditable, keep uniqExact but only after pre-aggregation has cut the input.
4. Enable spill. max_bytes_before_external_group_by = 12000000000 (about 40% of the 30 GiB limit) and the matching max_bytes_before_external_sort. Remember that a query can have a spilling group-by and a spilling sort at the same time, so do not hand both operators the full budget.
5. Lower max_threads for this query (e.g. 8): each thread keeps its own aggregation buffers, so fewer threads means a lower peak.

Spilling costs latency, which is acceptable for a nightly batch; the goal is to finish reliably, not quickly.

A read-to-output ratio like this means columns or partitions are being touched that never reach the result. Diagnose before changing anything: system.query_log exposes read_rows, read_bytes and read_columns per query, and EXPLAIN indexes = 1 shows which partitions and granules were selected.

Typical culprits, in the order worth checking:

• Wide reads from a subquery or view. A SELECT * in an inner query materialises url_raw, referrer, headers and other columns that the outer query discards. List columns explicitly everywhere.
• Filter placement. A date predicate hiding inside HAVING, a join ON clause, or a WHERE above an aggregation cannot prune partitions. It must be a top-level WHERE/PREWHERE on event_time for monthly partition pruning to work.
• String cardinality. page_url with 45,000 distinct values across billions of rows should be LowCardinality(String); that cuts both bytes read and the memory of the hash table holding the group keys.
• Repeated dimension reads. pages is 12M rows and nearly static. Loading page_id -> page_url into a Dictionary and using dictGet removes the join entirely and removes the 12M-row hash table from the memory budget.
• No pre-aggregation. If the report only needs daily uniques, an AggregatingMergeTree keyed by (day, page_id) with uniqCombinedState(user_id) makes the 30-day query merge 30 states per page instead of scanning raw events. Sketch states are composable, so uniqCombinedMerge over the month is consistent with the daily numbers.

Start with column lists, filter placement and LowCardinality - cheap, reversible, and usually most of the win - then re-measure before committing to a materialized view.

pageviews is Distributed, so the join executes on each shard. What each shard sees for pages determines correctness.

• pages is a local table replicated on every node (your case). Each shard joins its own slice of pageviews to the full pages table. Results are correct. The cost is that the 12M-row hash table is built three times, once per shard, which matters a lot when you are already near the memory limit - another argument for replacing the join with a dictionary lookup.
• pages as a Distributed table. Then each shard would join only against its own third of the dimension and silently drop matches. This is the case where GLOBAL JOIN is required: the right-hand side is built once on the initiator and shipped to every shard. distributed_product_mode (default deny) governs related IN/JOIN subquery shapes and will reject some of them outright.
• Alternatives to shipping. join_algorithm = 'direct' with a Dictionary turns the join into a local key-value lookup on every node. grace_hash and partial_merge spill to disk when the build side cannot fit in RAM; partial_merge requires sorted inputs, which your ORDER BY does not provide on page_id.

Practical checks: run EXPLAIN on the distributed query to see which side is shipped, confirm filters are pushed down to shards rather than applied on the initiator after the join, and verify shard-level row counts sum to the expected total. Given the memory profile, moving page_id -> page_url into a dictionary is the change that removes the join from the critical path altogether.