Setting up Citus

Citus is a sharding solution for PostgreSQL, and most deployments start from prebuilt containers that already bundle a working version of the extension. Container images skip most of the installation and configuration work, and manual setup is only needed when there is a reason to understand what runs underneath.

Container path

The quickest route to a running cluster is a Citus Docker container. A single command produces a working setup in minutes.

PgSQL

1

2

3

4

5

6

7

8

# run PostgreSQL with Citus on port 5500

docker run -d --name citus -p 5500:5432 -e POSTGRES_PASSWORD=mypassword citusdata/citus

# connect using psql within the Docker container

docker exec -it citus psql -U postgres

# or, connect using local psql

psql -U postgres -d postgres -h localhost -p 5500

Manual path

Installing the binaries by hand requires loading the Citus library at server startup. That means adding the citus extension to shared_preload_libraries in postgresql.conf:

PgSQL

1

2

shared_preload_libraries = 'citus,pg_stat_statements'  

# (change requires restart)

Adding pg_stat_statements to the same variable is useful for monitoring performance in detail. The ordering constraint matters: citus must come first in the list, or the server will refuse to start.

Coordinator and workers

Once the library is loaded, the cluster needs a set of database instances. A typical layout uses one coordinator plus worker nodes:

  • A coordinator: localhost, port 5432
  • 4 database worker nodes: localhost, ports 6001 - 6004

These instances are empty apart from the configuration above. The coordinator is registered first, then the workers are attached.

PgSQL

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

postgres=# SELECT citus_set_coordinator_host('localhost', 5432);

citus_set_coordinator_host

----------------------------

(1 row)

postgres=# SELECT citus_add_node('localhost', 6001);

citus_add_node

----------------

              2

(1 row)

postgres=# SELECT citus_add_node('localhost', 6002);

citus_add_node

----------------

              3

(1 row)

postgres=# SELECT citus_add_node('localhost', 6003);

citus_add_node

----------------

              4

(1 row)

postgres=# SELECT citus_add_node('localhost', 6004);

citus_add_node

----------------

              5

(1 row)

Workers are not for direct use

Shards are storage units, not endpoints for application traffic; operations are not supposed to run against them directly. Citus enforces this and returns an error when a client tries:

PgSQL

1

2

3

postgres=# CREATE EXTENSION pg_stat_statements ;

ERROR:  operation is not allowed on this node

HINT:  Connect to the coordinator and run it again.

Centralising access through the coordinator keeps shards consistent and prevents them from drifting out of control, which is what keeps the cluster healthy.

Creating and distributing the table

Creating the relation itself is an ordinary CREATE TABLE. That alone does not shard anything: Citus has to be told to distribute the table and to use a specific column as the distribution key. In this example the key column is called shard_key and lives in the table.

PgSQL

1

2

3

4

5

6

7

8

9

10

11

postgres=# CREATE TABLE t_shard (

id          serial,

shard_key   int,

n           int,

placeholder char(100) DEFAULT 'a');

CREATE TABLE

postgres=# SELECT create_distributed_table('t_shard', 'shard_key');

create_distributed_table

--------------------------

(1 row)

Data can then be sent to the table the normal way, and Citus handles placement automatically:

PgSQL

1

2

3

4

5

6

7

postgres=# timing

Timing is on.

postgres=# INSERT INTO t_shard (shard_key, n)

SELECT id % 16, random()*100000

FROM generate_series(1, 5000000) AS id;

INSERT 0 5000000

Time: 3443.606 ms (00:03.444)

On a local desktop machine, 5 million rows were inserted in barely more than 3 seconds — parallelism does appear to pay off here.

Table sizes need a caveat. Reporting the relation with the usual tooling is misleading:

PgSQL

1

2

3

4

5

6

7

8

9

postgres=# d+

                                          List of relations

Schema |      Name      |   Type   | Owner | Persistence | Access method |    Size    | …

--------+----------------+----------+-------+-------------+---------------+------------+ …

public | citus_schemas  | view     | hs    | permanent   |               | 0 bytes    |

public | citus_tables   | view     | hs    | permanent   |               | 0 bytes    |

public | t_shard        | table    | hs    | permanent   | heap          | 0 bytes    |

public | t_shard_id_seq | sequence | hs    | permanent   |               | 8192 bytes |

(4 rows)

There is no table “size” on the PostgreSQL side because Citus storage is in play. Real figures come from the Citus metadata:

PgSQL

1

2

3

4

5

6

7

8

9

10

11

12

postgres=# x

Expanded display is on.

postgres=# SELECT * FROM citus_tables ;

-[ RECORD 1 ]-------+------------

table_name          | t_shard

citus_table_type    | distributed

distribution_column | shard_key

colocation_id       | 1

table_size          | 711 MB

shard_count         | 32

table_owner         | hs

access_method       | heap

This matters for monitoring, since many tools rely on the standard PostgreSQL size functions and therefore need to be made aware of sharded tables. Users of pgwatch can add metrics that account for it.

Query execution

An unqualified aggregate shows how work is spread. Running a count(*) against a distributed table reveals the execution plan:

PgSQL

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

postgres=# explain (analyze, buffers, timing) SELECT count(*)

FROM t_shard;

                                                                                 QUERY PLAN                                                                                  

-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------

Aggregate  (cost=250.00..250.02 rows=1 width=8) (actual time=520.360..520.361 rows=1 loops=1)

   Buffers: shared hit=93

   ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=100000 width=8) (actual time=520.347..520.350 rows=32 loops=1)

         Task Count: 32

         Tuple data received from nodes: 256 bytes

         Tasks Shown: One of 32

         ->  Task

               Tuple data received from node: 8 bytes

               Node: host=localhost port=6001 dbname=postgres

               ->  Finalize Aggregate  (cost=22929.03..22929.04 rows=1 width=8) (actual time=453.723..454.161 rows=1 loops=1)

                     Buffers: shared hit=2336 read=14710

                     ->  Gather  (cost=22928.81..22929.02 rows=2 width=8) (actual time=453.632..454.157 rows=3 loops=1)

                           Workers Planned: 2

                           Workers Launched: 2

                           Buffers: shared hit=2336 read=14710

                           ->  Partial Aggregate  (cost=21928.81..21928.82 rows=1 width=8) (actual time=448.045..448.047 rows=1 loops=3)

                                 Buffers: shared hit=2336 read=14710

                                 ->  Parallel Seq Scan on t_shard_102016 t_shard  (cost=0.00..20952.25 rows=390625 width=0) (actual time=0.094..434.632 rows=312500 loops=3)

                                       Buffers: shared hit=2336 read=14710

                   Planning Time: 0.127 ms

                   Execution Time: 454.193 ms

         Buffers: shared hit=93

Planning:

   Buffers: shared hit=349

Planning Time: 5.199 ms

Execution Time: 520.431 ms

(26 rows)

PostgreSQL distributes the query and sends it to every shard. The plan shows the operations for one shard only rather than all of them; since the shards are created equal, that is sufficient for a picture of the work.

Filtering on a specific shard_key value is different: only a single shard is addressed.

PgSQL

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

postgres=# explain (analyze, buffers, timing) SELECT count(*)

FROM t_shard

WHERE shard_key = 2;

                                                                             QUERY PLAN                                                                              

---------------------------------------------------------------------------------------------------------------------------------------------------------------------

Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=0 width=0) (actual time=33.874..33.875 rows=1 loops=1)

   Task Count: 1

   Tuple data received from nodes: 8 bytes

   Tasks Shown: All

   ->  Task

         Tuple data received from node: 8 bytes

         Node: host=localhost port=6001 dbname=postgres

         ->  Finalize Aggregate  (cost=8635.34..8635.35 rows=1 width=8) (actual time=30.490..31.309 rows=1 loops=1)

               Buffers: shared hit=2208 read=3474

               ->  Gather  (cost=8635.12..8635.33 rows=2 width=8) (actual time=30.407..31.303 rows=3 loops=1)

                     Workers Planned: 2

                     Workers Launched: 2

                     Buffers: shared hit=2208 read=3474

                     ->  Partial Aggregate  (cost=7635.12..7635.13 rows=1 width=8) (actual time=26.023..26.024 rows=1 loops=3)

                           Buffers: shared hit=2208 read=3474

                           ->  Parallel Seq Scan on t_shard_102032 t_shard  (cost=0.00..7309.60 rows=130208 width=0) (actual time=0.420..19.400 rows=104167 loops=3)

                                 Filter: (shard_key = 2)

                                 Buffers: shared hit=2208 read=3474

             Planning Time: 0.516 ms

             Execution Time: 31.384 ms

Planning Time: 0.146 ms

Execution Time: 33.903 ms

(22 rows)

The “Node” entry in the plan points to the instance on port 6001, which confirms that not all shards are scanned — a significant benefit of distribution.

Sharding is not indexing

Sharding is often treated as a general performance fix. It is one ingredient, not a replacement for indexing or for thinking about the query. Take this query:

PgSQL

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

postgres=# explain (analyze, buffers, timing) SELECT count(*)

FROM t_shard

WHERE id = 10;

                                                                          QUERY PLAN                                                                          

---------------------------------------------------------------------------------------------------------------------------------------------------------------

Aggregate  (cost=250.00..250.02 rows=1 width=8) (actual time=541.532..541.534 rows=1 loops=1)

   Buffers: shared hit=3

   ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=100000 width=8) (actual time=540.910..540.914 rows=32 loops=1)

         Task Count: 32

         Tuple data received from nodes: 256 bytes

         Tasks Shown: One of 32

         ->  Task

               Tuple data received from node: 8 bytes

               Node: host=localhost port=6001 dbname=postgres

               ->  Aggregate  (cost=22928.91..22928.92 rows=1 width=8) (actual time=463.883..464.342 rows=1 loops=1)

                     Buffers: shared hit=2528 read=14518

                     ->  Gather  (cost=1000.00..22928.91 rows=1 width=0) (actual time=463.881..464.339 rows=0 loops=1)

                           Workers Planned: 2

                           Workers Launched: 2

                           Buffers: shared hit=2528 read=14518

                           ->  Parallel Seq Scan on t_shard_102016 t_shard  (cost=0.00..21928.81 rows=1 width=0) (actual time=456.989..456.990 rows=0 loops=3)

                                 Filter: (id = 10)

                                 Rows Removed by Filter: 312500

                                 Buffers: shared hit=2528 read=14518

                   Planning Time: 0.267 ms

                   Execution Time: 464.378 ms

         Buffers: shared hit=3

Planning:

   Buffers: shared hit=28

Planning Time: 4.432 ms

Execution Time: 542.064 ms

(26 rows)

Citus engages the sharding machinery, queries in parallel and returns a result — but it has to go through all of the data. Adding hardware to a query with that shape makes things worse, not better. The fix is an index:

PgSQL

1

2

postgres=# CREATE INDEX ON t_shard (id);

CREATE INDEX

With the index in place, performance improves by orders of magnitude:

PgSQL

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

postgres=# explain (analyze, buffers, timing) SELECT count(*) FROM t_shard WHERE id = 10;

                                                                                QUERY PLAN                                                                                

---------------------------------------------------------------------------------------------------------------------------------------------------------------------------

Aggregate  (cost=250.00..250.02 rows=1 width=8) (actual time=8.795..8.796 rows=1 loops=1)

   ->  Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=100000 width=8) (actual time=8.780..8.783 rows=32 loops=1)

         Task Count: 32

         Tuple data received from nodes: 256 bytes

         Tasks Shown: One of 32

         ->  Task

               Tuple data received from node: 8 bytes

               Node: host=localhost port=6001 dbname=postgres

               ->  Aggregate  (cost=4.44..4.45 rows=1 width=8) (actual time=0.053..0.053 rows=1 loops=1)

                     Buffers: shared read=3

                     ->  Index Only Scan using t_shard_id_idx_102008 on t_shard_102008 t_shard  (cost=0.42..4.44 rows=1 width=0) (actual time=0.049..0.049 rows=0 loops=1)

                           Index Cond: (id = 10)

                           Heap Fetches: 0

                           Buffers: shared read=3

                   Planning Time: 0.466 ms

                   Execution Time: 0.479 ms

Planning:

   Buffers: shared hit=117 read=1

Planning Time: 2.537 ms

Execution Time: 8.875 ms

(20 rows)

That is orders of magnitude faster than without an index, and also orders of magnitude slower than the same work on a local, non-sharded database. There is no magic solution to performance — not even sharding.

For background on sharding, indexing and performance, the following posts are worth reading:

  • Other blog posts on scalability, for anyone new to sharding and Citus on PostgreSQL.
  • Choosing bigint or uuid for an auto-generated primary key.
  • New target_session_attrs settings for high availability and scaling in PostgreSQL v14.
  • PostgreSQL scaling advice for 2021.

In order to receive regular updates on important changes in PostgreSQL, subscribe to the newsletter, or follow on Facebook or LinkedIn.