The cost of ignoring data locality
Sharding is frequently presented as the answer to PostgreSQL scalability, but it only works when you know what it does to your data. Among the most common mistakes is neglecting data locality — a concern that matters in many areas of IT and is decisive once a database is sharded. Citus, one of the more sophisticated sharding solutions available for PostgreSQL, supports both analytics and OLTP workloads and is offered on-premise or through Microsoft Azure.
Setting up a sharded example
The demonstration starts with two simple tables, customers and sales:
|
1 2 3 4 5 6 7 8 9 10 11 |
postgres=# CREATE TABLE t_customer ( id int, name text ); CREATE TABLE postgres=# CREATE TABLE t_sales ( id int, customer_id int, whatever text ); CREATE TABLE |
Analyzing this model typically means joining customers against sales. The tables are then distributed and populated:
|
1 2 3 4 5 6 7 8 9 10 11 |
postgres=# SELECT create_distributed_table('t_customer', 'id'); create_distributed_table -------------------------- (1 row) postgres=# SELECT create_distributed_table('t_sales', 'id'); create_distributed_table -------------------------- (1 row) |
Note that the table is sharded on id, which is not the join criterion.
|
1 2 3 4 5 6 7 8 9 |
postgres=# INSERT INTO t_customer SELECT *, 'dummy' FROM generate_series(1, 1000000); INSERT 0 1000000 postgres=# INSERT INTO t_sales SELECT id, random()*100000, 'dummy' FROM generate_series(1, 1000000) AS id; INSERT 0 1000000 |
What happens when the query runs
A straightforward join-and-count query across the two sharded tables does not execute as written:
|
1 2 3 4 5 |
postgres=# SELECT count(*) FROM t_customer AS c, t_sales AS s WHERE c.id = s.customer_id; ERROR: the query contains a join that requires repartitioning HINT: Set citus.enable_repartition_joins to on to enable repartitioning |
Citus rejects the query by default because the join does not use the sharding criterion. The reasoning is mechanical: when both rows live on the same machine, PostgreSQL can join them locally in RAM, which is efficient. Once the data has to traverse an expensive network stack, the cost grows by orders of magnitude, and on large data sets the approach becomes impractical. Returning an error is preferable to silently steering the user into a serious performance problem.
Forcing the join is possible through a locally set configuration variable, although this is not something to do in production on large data sets:
|
1 2 3 4 5 6 7 8 9 |
postgres=# SET citus.enable_repartition_joins = on; SET postgres=# SELECT count(*) FROM t_customer AS c, t_sales AS s WHERE c.id = s.customer_id; count -------- 999993 (1 row) |
The query returns in reasonable time here only because the test data set is small. On large, non-colocated data sets the query may never finish.
The execution plan shows what is actually happening:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
postgres=# explain analyze SELECT count(*) FROM t_customer AS c, t_sales AS s WHERE c.id = s.customer_id; QUERY PLAN ---------------------------------------------------------------------------------------------------------------------------- Aggregate (cost=250.00..250.02 rows=1 width=8) (actual time=1283.486..1283.486 rows=1 loops=1) -> Custom Scan (Citus Adaptive) (cost=0.00..0.00 rows=100000 width=8) (actual time=1283.478..1283.480 rows=20 loops=1) Task Count: 20 Tuple data received from nodes: 160 bytes Tasks Shown: None, not supported for re-partition queries -> MapMergeJob Map Task Count: 32 Merge Task Count: 20 -> MapMergeJob Map Task Count: 32 Merge Task Count: 20 Planning Time: 10.282 ms Execution Time: 1283.670 ms (13 rows) |
The run takes roughly 1.3 seconds. If the shards sit on separate machines rather than separate ports, that time can easily multiply.
Colocating the tables
The fix is data locality: rows that must be joined belong in the same shard. Data should not be moved between shards, and cross-shard operations should be avoided wherever possible. In practice, both tables have to use the same sharding criterion:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
postgres=# CREATE TABLE t_sales_proper AS SELECT * FROM t_sales; SELECT 1000000 postgres=# SELECT create_distributed_table( 't_sales_proper', 'customer_id', colocate_with => 't_customer' ); NOTICE: Copying data from local table... NOTICE: copying the data has completed DETAIL: The local data in the table is no longer visible, but is still on disk. HINT: To remove the local data, run: SELECT truncate_local_data_after_distributing_table($public.t_sales_proper$) create_distributed_table -------------------------- (1 row) |
Citus warns that a local table with all the data now exists. For safety, create_distributed_table does not remove local data, so the local copy has to be truncated to avoid storing everything twice:
|
1 2 3 4 5 |
postgres=# SELECT truncate_local_data_after_distributing_table($public.t_sales_proper$); truncate_local_data_after_distributing_table ---------------------------------------------- (1 row) |
Keeping data locally is generally a bad idea, since space can run out. Sharding and Citus are about scalability, and centralizing data runs against that goal on both space and other limitations.
With the data properly sharded, the query can be run again:
|
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 SELECT count(*) FROM t_customer AS c, t_sales_proper AS s WHERE c.id = s.customer_id; QUERY PLAN ----------------------------------------------------------------------------------------------------------------------------------------------------------- Aggregate (cost=250.00..250.02 rows=1 width=8) (actual time=98.915..98.915 rows=1 loops=1) -> Custom Scan (Citus Adaptive) (cost=0.00..0.00 rows=100000 width=8) (actual time=98.907..98.909 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=6003 dbname=postgres -> Aggregate (cost=1851.57..1851.58 rows=1 width=8) (actual time=25.905..25.907 rows=1 loops=1) -> Hash Join (cost=858.76..1773.21 rows=31345 width=0) (actual time=8.929..23.526 rows=31345 loops=1) Hash Cond: (s.customer_id = c.id) -> Seq Scan on t_sales_proper_102142 s (cost=0.00..483.45 rows=31345 width=4) (actual time=0.007..2.387 rows=31345 loops=1) -> Hash (cost=474.45..474.45 rows=30745 width=4) (actual time=8.863..8.863 rows=30745 loops=1) Buckets: 32768 Batches: 1 Memory Usage: 1337kB -> Seq Scan on t_customer_102078 c (cost=0.00..474.45 rows=30745 width=4) (actual time=0.005..3.701 rows=30745 loops=1) Planning Time: 0.138 ms Execution Time: 26.038 ms Planning Time: 2.678 ms Execution Time: 98.964 ms |
The result is a query that is 15 times faster than before — and in real-world cases the gap is often larger still.
