Top-N and Percentile Queries Across Shards

Ranked and quantile questions are the two aggregate shapes that break naive fan-out. Both return plausible numbers when implemented wrongly, which is why they survive review and fail an audit. This guide gives the correct over-fetch rule for top-N, replaces deep pagination with keyset traversal, and merges sketches for percentiles. It extends Cross-Shard Aggregation Patterns inside Cross-Partition Querying & Aggregation Strategies.

Prerequisites

• A fan-out coordinator that can issue the same query to every shard and merge results
• PostgreSQL 13+; the tdigest extension on each shard for percentile merging
• A stable tiebreaker column so ordering is total, not merely partial
• An agreed accuracy requirement per query — exact or bounded-error
• Latency budgets for the fan-out, since these shapes are the slowest ones

Step 1 — Apply the over-fetch rule for top-N

The rule has one line and one reason:

-- each shard, for a global LIMIT 10
SELECT customer_id, sum(amount_cents) AS spend
FROM   orders
WHERE  placed_at >= '2026-08-01' AND placed_at < '2026-09-01'
GROUP  BY customer_id
ORDER  BY spend DESC, customer_id      -- tiebreaker makes the order total
LIMIT  10;                             -- N, never fewer

# coordinator: merge the per-shard lists and cut once
import heapq

def top_n(shard_results, n):
    merged = heapq.merge(*shard_results, key=lambda r: (-r.spend, r.customer_id))
    return list(itertools.islice(merged, n))

The reason to fetch N from every shard is the worst case: all ten global winners could be on one shard. Fetching nine from each is correct almost always and wrong exactly when the data is skewed, which is when someone is looking.

Operational note: The tiebreaker is not cosmetic. Without a total order, two shards can return the same value in different orders and the merge becomes non-deterministic between runs — which looks like data instability to whoever is reading the dashboard.

DBA tip: Push the GROUP BY down. Sending raw order rows to the coordinator to aggregate there moves orders of magnitude more data for the same answer.

Step 2 — Replace OFFSET with a cursor

-- each shard: rows strictly after the cursor, in the same total order
SELECT customer_id, spend
FROM   customer_spend_mv
WHERE  (spend, customer_id) < (%(last_spend)s, %(last_customer_id)s)
ORDER  BY spend DESC, customer_id
LIMIT  20;

# the cursor is the last row of the previous page, opaque to the client
def next_page(coordinator, cursor, n=20):
    results = coordinator.fan_out(SQL, cursor=cursor, limit=n)
    page = merge_and_cut(results, n)
    return page, (page[-1].spend, page[-1].customer_id) if page else None

Operational note: Row-value comparison ((a, b) < (x, y)) is a single index-friendly predicate in PostgreSQL, provided an index exists on (spend DESC, customer_id). Splitting it into spend < x OR (spend = x AND id < y) usually loses the index.

DBA tip: Keyset pagination cannot jump to an arbitrary page number. If the product requires page numbers, cap the reachable depth — nobody paginates to page 500 by hand, but a crawler will.

Step 3 — Merge sketches for percentiles

Averaging per-shard percentiles is wrong; merging digests is right:

-- on each shard: return sketch state, not a number
CREATE EXTENSION IF NOT EXISTS tdigest;

SELECT tdigest(response_ms, 100) AS digest
FROM   request_log
WHERE  logged_at >= now() - interval '1 hour';

# coordinator: merge the digests, then ask the merged sketch
from tdigest import TDigest

def p99_across_shards(digests):
    merged = TDigest()
    for d in digests:
        merged = merged + d            # merging is associative and commutative
    return merged.percentile(99)

Operational note: Store the digest state rather than the percentile in any rollup table. A stored p99 can only answer the question it was computed for; stored digest state answers p50, p95 and p99.9 later, and can be re-merged across arbitrary groupings.

DBA tip: Size the digest compression parameter for tail accuracy, not for size. The default of 100 is accurate to within a fraction of a percent at p99 and costs a few kilobytes — increasing it buys precision you will not measure.

Step 4 — Bound the fan-out for these shapes specifically

Ranked and quantile queries are the ones most worth pre-aggregating:

-- per-shard hourly rollup: sketch state plus the top slice
CREATE TABLE spend_hourly (
    shard_name  text        NOT NULL,
    bucket_ts   timestamptz NOT NULL,
    top_100     jsonb       NOT NULL,      -- customer_id/spend pairs
    latency_td  tdigest     NOT NULL,
    PRIMARY KEY (shard_name, bucket_ts)
);

The dashboard then merges 16 small rows instead of fanning out to 16 databases, and the fan-out runs once an hour on a schedule rather than on every page load.

Operational note: Keeping the top 100 per shard makes a global top 10 or top 50 exact from the rollup. A global top 200 would not be, so the rollup’s slice size must exceed the largest N the product exposes.

SRE tip: Log the requested N alongside the rollup’s slice size. The day someone adds a “top 500” report, that log line is what prevents a silently wrong answer.

Verification

Test correctness against a known-skewed dataset, which is where the wrong implementations fail:

def test_top_n_with_all_winners_on_one_shard():
    seed_shard("shard_0", [(f"c{i}", 1_000_000 - i) for i in range(10)])   # all winners
    for s in ("shard_1", "shard_2", "shard_3"):
        seed_shard(s, [(f"x{i}", 100 + i) for i in range(50)])
    result = top_n_across_shards(n=10)
    assert [r.customer_id for r in result] == [f"c{i}" for i in range(10)]

def test_percentile_merge_matches_exact_within_tolerance():
    exact = numpy.percentile(all_values, 99)
    merged = p99_across_shards(per_shard_digests)
    assert abs(merged - exact) / exact < 0.005

test_top_n_with_all_winners_on_one_shard PASSED
test_percentile_merge_matches_exact_within_tolerance PASSED  (merged 418.2 vs exact 417.9)

Failure mode table

FAQ

Related

• Cross-Shard Aggregation Patterns — the parent topic, including which aggregates merge trivially
• Optimizing Cross-Partition Aggregations with Materialized Views — the rollup layer these queries should read from
• Application-Level Sharding Logic — bounded fan-out, deadlines and partial results