Using Citus for Distributed PostgreSQL: SQL Reference for Querying Distributed Tables

Aggregation Functions

Citus extends PostgreSQL and parallelises SELECT queries that contain filters, GROUP BY, ORDER BY and JOINs. The query is split into fragments, dispatched to workers, executed, and the results are merged (optionally sorted) before being returned.

Citus supports the full set of PostgreSQL aggregates, including user‑defined ones, using three execution strategies in priority order:

When the aggregate groups by the distribution column, the whole query is pushed down to each worker, allowing full parallel execution. Custom aggregates must be installed on the workers.

If the aggregate does not group by the distribution column, Citus can still optimise certain aggregates (e.g., sum(), avg(), count(distinct)) by performing a partial aggregation on workers and completing the final step on the coordinator.

As a fallback, Citus pulls all rows from workers to the coordinator and performs the aggregation there. This may cause network overhead and coordinator resource exhaustion; the fallback can be disabled.

Small query changes can switch execution modes. For example, sum(x) may run distributed while sum(distinct x) forces coordinator aggregation.

SELECT sum(value1), sum(DISTINCT value2) FROM distributed_table;

To prevent accidental coordinator aggregation, set the GUC:

SET citus.coordinator_aggregation_strategy TO 'disabled';

Count (Distinct) Aggregation

If the distinct column is the distribution column, Citus pushes the count(distinct) down to workers. Otherwise each worker runs a SELECT DISTINCT and the coordinator aggregates the results. Multiple distinct counts in a single query can generate large cross‑product transfers.

-- multiple distinct counts in one query tend to be slow
SELECT count(DISTINCT a), count(DISTINCT b), count(DISTINCT c)
FROM table_abc;

HyperLogLog Columns

When data is stored in an HLL column, merge sketches with hll_union_agg(hll_column).

Estimating Top‑N Items

The open‑source TopN extension provides an approximate top‑N result stored as JSON and supports incremental updates. Repository: https://github.com/citusdata/postgresql-topn

Basic Operations

Updating a JSON top‑N counter:

select topn_add('{}', 'a');
-- => {"a": 1}

select topn_add(topn_add('{}', 'a'), 'a');
-- => {"a": 2}

Aggregating many values with topn_add_agg:

CREATE EXTENSION tablefunc;
SELECT topn_add_agg(floor(abs(i))::text)
FROM normal_rand(1000, 5, 0.7) i;
-- => {"2": 1, "3": 74, "4": 420, "5": 425, "6": 77, "7": 3}

The number of distinct values kept can be limited with the GUC topn.number_of_counters (default 1000).

Real‑World Example

Download a 2000‑year Amazon review dataset and ingest it into a distributed table:

curl -L https://examples.citusdata.com/customer_reviews_2000.csv.gz | \
  gunzip > reviews.csv

CREATE TABLE customer_reviews (
    customer_id TEXT,
    review_date DATE,
    review_rating INTEGER,
    review_votes INTEGER,
    review_helpful_votes INTEGER,
    product_id CHAR(10),
    product_title TEXT,
    product_sales_rank BIGINT,
    product_group TEXT,
    product_category TEXT,
    product_subcategory TEXT,
    similar_product_ids CHAR(10)[]
);
SELECT create_distributed_table('customer_reviews', 'product_id');
\COPY customer_reviews FROM 'reviews.csv' WITH CSV;

Create the TopN extension and a reference table to materialise daily aggregates:

CREATE EXTENSION topn;
CREATE TABLE reviews_by_day (
  review_date DATE UNIQUE,
  agg_data JSONB
);
SELECT create_reference_table('reviews_by_day');
INSERT INTO reviews_by_day
  SELECT review_date, topn_add_agg(product_id)
  FROM customer_reviews
  GROUP BY review_date;

Query the most reviewed product per day for the first five days:

SELECT review_date, (topn(agg_data, 1)).*
FROM reviews_by_day
ORDER BY review_date
LIMIT 5;

┌─────────────┬────────────┬───────────┐
│ review_date │    item    │ frequency │
├─────────────┼────────────┼───────────┤
│ 2000-01-01  │ 0939173344 │        12 │
│ 2000-01-02  │ B000050XY8 │        11 │
│ 2000-01-03  │ 0375404368 │        12 │
│ 2000-01-04  │ 0375408738 │        14 │
│ 2000-01-05  │ B00000J7J4 │        17 │
└─────────────┴────────────┴───────────┘

Merge a month’s data and list the top five products:

SELECT (topn(topn_union_agg(agg_data), 5)).*
FROM reviews_by_day
WHERE review_date >= '2000-01-01' AND review_date < '2000-02-01'
ORDER BY 2 DESC;

┌────────────┬───────────┐
│    item    │ frequency │
├────────────┼───────────┤
│ 0375404368 │       217 │
│ 0345417623 │       217 │
│ 0375404376 │       217 │
│ 0375408738 │       217 │
│ 043936213X │       204 │
└────────────┴───────────┘

Percentile Calculation

Exact percentiles require moving all rows to the coordinator. Approximate percentiles use sketch algorithms such as t‑digest, which run on workers and send only compressed summaries.

Install the t‑digest extension (https://github.com/tvondra/tdigest) on all PostgreSQL nodes: CREATE EXTENSION tdigest; When using t‑digest aggregates, Citus pushes the partial computation to workers. Accuracy can be tuned via the compression parameter, trading precision against the amount of data transferred between workers and the coordinator.

Limit Push‑Down

Citus pushes LIMIT clauses to worker shards when possible, reducing network traffic. For queries that require sorting across shards, Citus may need to fetch all rows, making LIMIT less efficient. Approximate LIMIT can be enabled with the GUC citus.limit_clause_row_fetch_count and tuned by adjusting the row‑fetch count.

SET citus.limit_clause_row_fetch_count TO 10000;

Distributed Table Views

All PostgreSQL view syntax works on distributed tables. Materialised views are stored locally on the coordinator.

Joins

Citus supports arbitrary equi‑JOINs between any number of tables, choosing the optimal join order based on distribution to minimise data movement.

Co‑located Joins

When two tables share the same distribution column, a co‑located join avoids data movement and is the most efficient for large tables.

Ensure tables are sharded into the same number of shards and that distribution columns have exactly matching types; mismatched types (e.g., int vs bigint ) can cause issues.

Reference Table Joins

Reference tables are fully replicated on all workers and act as dimension tables. Joins with reference tables are executed locally on each worker, similar to co‑located joins but without a specific distribution key.

Repartition Joins

If a join key is not the distribution column, Citus can dynamically repartition one side of the join. The optimizer decides which table to repartition based on size and distribution columns, ensuring only relevant shards are shuffled. Co‑located joins are generally preferred because repartition joins incur a shuffle.

More Resources

For additional SQL reference topics such as creating and modifying distributed tables (DDL) and data ingestion (DML), see the Citus documentation site.