Cross-Shard Aggregation Patterns

Cross-shard aggregation is one of the operationally demanding corners of the broader Cross-Partition Querying & Aggregation Strategies space. Where single-shard queries are fast and local, aggregations that span every partition require a coordinator to orchestrate parallel dispatch, collect partial results, and merge them without exhausting memory or blocking on a slow shard. This page covers the architecture, implementation, and failure modes of that full pipeline — from scatter-gather wiring through pre-aggregation pushdown to streaming analytics. For the routing layer that sits upstream of aggregation, see Proxy Routing Architectures.

Problem Framing

Consider a 2 TB events table sharded across 16 PostgreSQL nodes by tenant_id. A billing job needs SUM(usage_bytes) GROUP BY tenant_id, DATE_TRUNC('day', created_at) for the past 30 days. If the coordinator naively fans out a full-table query to every shard and buffers all rows before grouping, it transfers hundreds of gigabytes across the network and risks coordinator OOM. With a shard timeout of 2 s, a single slow node stalls the entire merge until the timeout fires — at which point the result is either dropped or returned as approximate. The patterns below address each of these failure vectors in turn.

Architecture Overview

The diagram below shows the full scatter-gather pipeline: a coordinator decomposes the query, dispatches in parallel to each shard, collects partial results (already pre-aggregated on the shard), merges them, and returns a final result with an accuracy flag.

Implementation Walkthrough

Step 1 — Decompose the Query and Prune Shards

Before dispatching, identify which shards hold relevant data. If the query includes a tenant_id filter, only send to the shard(s) that own those tenants. Full scatter is only correct when no shard key is present in the predicate.

-- On each shard: push the GROUP BY down before shipping rows
SELECT
  tenant_id,
  DATE_TRUNC('day', created_at) AS day,
  SUM(usage_bytes)              AS partial_sum,
  COUNT(*)                      AS partial_count
FROM   events
WHERE  created_at >= NOW() - INTERVAL '30 days'
GROUP  BY 1, 2;

Returning partial_sum and partial_count separately lets the coordinator compute a correct weighted average (SUM(partial_sum) / SUM(partial_count)) without raw rows crossing the network.

Step 2 — Dispatch in Parallel with Per-Shard Timeouts

Use Promise.allSettled (or its equivalent in your language) so a single slow shard cannot block the merge indefinitely.

async function executeCrossShardAggregation(query, shards) {
  const partials = await Promise.allSettled(
    shards.map(s => dispatchToShard(s, query, { timeoutMs: 2000 }))
  );

  const successful = partials
    .filter(p => p.status === 'fulfilled')
    .map(p => p.value);

  const failedCount = partials.filter(p => p.status === 'rejected').length;

  return {
    result:       mergePartials(successful),
    accuracy:     failedCount > 0 ? 'approximate' : 'exact',
    failedShards: failedCount
  };
}

failedShards flows directly into your observability pipeline. Wire it to an alert so on-call knows when results are approximate before a downstream consumer flags a billing discrepancy.

Step 3 — Merge Partial Results on the Coordinator

The merge logic must be associative. SUM and COUNT are straightforward; AVG requires the count-weighted form; DISTINCT COUNT requires either exact distinct counts per shard (only correct when keys do not overlap) or a HyperLogLog sketch.

from functools import reduce

def merge_partials(partials: list[dict]) -> dict:
    """Merge pre-aggregated shard results for SUM and COUNT."""
    return reduce(
        lambda acc, p: {
            'total_sum':   acc['total_sum']   + p['partial_sum'],
            'total_count': acc['total_count'] + p['partial_count'],
        },
        partials,
        {'total_sum': 0, 'total_count': 0}
    )

Step 4 — Stream Results When Cardinality Is High

For high-cardinality GROUP BY (millions of groups), buffering all partial rows in coordinator memory before merging triggers OOM. Use a streaming merge instead: open a cursor on each shard, read rows in sorted order, and merge-sort across shards.

-- On each shard: open a named cursor so rows stream on FETCH
BEGIN;
DECLARE partial_cur CURSOR FOR
  SELECT tenant_id, day, SUM(usage_bytes) AS partial_sum
  FROM   events
  WHERE  created_at >= NOW() - INTERVAL '30 days'
  GROUP  BY 1, 2
  ORDER  BY tenant_id, day;

FETCH 500 FROM partial_cur;

The coordinator reads 500-row batches from all shards in lockstep, emitting merged rows as it goes. Peak coordinator memory stays proportional to the number of shards × batch size, not to total result cardinality.

Step 5 — Persist Expensive Aggregations as Materialized Views

For aggregations that run on a schedule (daily billing, hourly dashboards), avoid scatter-gather on every request. Pre-compute and cache the result. See Optimizing Cross-Partition Aggregations with Materialized Views for the full refresh workflow.

-- Citus: parallel materialized view across distributed shards
SET citus.max_parallel_tasks_per_job = 4;

CREATE MATERIALIZED VIEW mv_daily_usage AS
SELECT
  tenant_id,
  DATE_TRUNC('day', created_at) AS day,
  SUM(usage_bytes)              AS total_bytes
FROM   events
GROUP  BY 1, 2;

-- Refresh only the delta partition (last 2 days) to reduce compute
REFRESH MATERIALIZED VIEW CONCURRENTLY mv_daily_usage;

Enable enable_partition_pruning = on and jit = on on each shard node so the local aggregation step is JIT-compiled and only touches the relevant date partitions.

Configuration Reference

Operational Contrast

This scatter-gather model differs from the two adjacent approaches in the same section:

• Application-Level Sharding Logic embeds routing directly in the application process, eliminating a network hop but requiring every service to carry its own shard topology awareness. Aggregation logic then also lives in the application, making cross-team reuse harder.
• Proxy Routing Architectures centralise routing in a dedicated proxy (e.g. Pgpool, ProxySQL, or a custom HAProxy config) that can cache routing tables and connection pools. The proxy is the right place to colocate a coordinator when multiple services need the same aggregation paths, because the routing config lives in one place. For how to build that proxy layer, see Building a Custom Query Router with HAProxy.
• Federated Query Execution goes a step further: engines such as Trino and DuckDB treat heterogeneous data sources (Parquet, S3, JDBC) as first-class shards, handling the scatter-gather and merge internally. Choose federated execution when your data spans multiple storage systems; use the native scatter-gather patterns on this page when all shards share the same engine (PostgreSQL, MySQL, Citus).

Choosing Freshness Deliberately

Every cross-shard aggregate answers a question about data that is changing while the query runs, so “correct” is only meaningful relative to a stated freshness. Teams that skip this decision end up defending exact numbers that were never guaranteed, and paying scatter-gather prices for dashboards nobody refreshes.

Three freshness tiers cover almost every reporting need, and each maps to a different implementation:

The important discipline is to write the tier into the API contract rather than leaving it implicit. A response that carries "as_of": "2026-08-03T09:15:00Z" lets a caller decide whether the number is fresh enough, and it turns an argument about correctness into an inspection of a timestamp.

Reconciling the tiers

Systems that serve both a live dashboard and a monthly invoice inevitably run two paths over the same data, and the two will disagree. That is expected; what matters is that the disagreement is bounded and detectable. Run a scheduled reconciliation that compares the rollup against a full scatter-gather for a sample of keys:

-- run hourly; alert when the drift exceeds the tier's promise
WITH exact AS (
  SELECT tenant_id, sum(amount_cents) AS exact_total
  FROM   orders
  WHERE  created_at >= date_trunc('day', now())
  GROUP  BY tenant_id
)
SELECT r.tenant_id,
       r.rollup_total,
       e.exact_total,
       abs(r.rollup_total - e.exact_total) AS drift
FROM   rollup_daily r
JOIN   exact e USING (tenant_id)
WHERE  abs(r.rollup_total - e.exact_total) > 0
ORDER  BY drift DESC
LIMIT  20;

Persistent non-zero drift on the same tenant is almost always a refresh watermark that advanced past rows which arrived late — a row written with a timestamp earlier than the watermark it was committed under. The fix is to drive the delta refresh from a monotonic commit column rather than from the business timestamp, and to overlap the window by a safety margin.

When exactness is genuinely required

Financial totals, inventory counts and anything a regulator reads must come from the transactional tier, and that constrains the architecture upstream: those aggregates should be computed per shard and summed, never sampled or approximated, and the query must fail rather than return a partial result when a shard is unreachable. Encode that in the coordinator explicitly — a flag on the request that turns partial results from a degraded response into an error — so the same fan-out machinery can serve both a dashboard that tolerates a missing shard and an invoice run that must not.

The pattern to avoid is a single “get totals” endpoint used by both. It will eventually be called during a shard outage, and whichever behaviour it has chosen will be wrong for one of its two callers.

Monitoring and Observability

Track these signals to detect aggregation problems before they affect downstream consumers.

# Coordinator latency — alert when P95 exceeds 1.5 s
histogram_quantile(0.95,
  rate(scatter_gather_duration_seconds_bucket[5m])
) > 1.5

# Approximate-result rate — alert when more than 5 % of aggregations are partial
rate(scatter_gather_approximate_total[5m])
  / rate(scatter_gather_requests_total[5m]) > 0.05

# Partial result payload — detect unbounded memory growth
rate(coordinator_partial_result_bytes_total[5m]) > 104857600

Also instrument failedShards as a gauge so your on-call dashboard shows exactly which shard is timing out and for how long.

Failure Modes

Common Mistakes

• Full table scans on every shard for every aggregation. Not applying shard-key predicates means the coordinator fans out to all shards even when only a small tenant cohort is needed. Add tenant_id to the WHERE clause whenever the caller knows the scope, and let the routing layer prune shards before dispatch.

• Treating DISTINCT COUNT like SUM. Summing per-shard distinct counts double-counts keys that appear on multiple shards. Use HyperLogLog sketches (e.g. pg_hll) for approximate distinct counts, or reroute the query through a single shard if exact counts are required.

• No accuracy flag on the response. When a shard times out and is excluded, the merge result is incorrect but looks exact. Downstream services (billing, reporting) must know the result is approximate so they can hold it for re-computation rather than committing it.

• Shared coordinator for OLTP and OLAP traffic. Running scatter-gather aggregations on the same coordinator that handles OLTP writes creates CPU contention and unpredictable latency. Route aggregation queries to a dedicated read-replica pool and set statement_timeout independently on each pool.

FAQ

Related

• Cross-Partition Querying & Aggregation Strategies — parent section covering the full query execution and routing landscape
• Optimizing Cross-Partition Aggregations with Materialized Views — caching strategies and incremental refresh for expensive aggregations
• Proxy Routing Architectures — centralized routing layer that complements coordinator placement
• Application-Level Sharding Logic — embedding shard routing directly in the application process
• Federated Query Execution — cross-engine aggregation with Trino, DuckDB, and heterogeneous data sources