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).

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