Split the Users Table

Every lab includes hints and full solutions, so don’t worry if you’re not confident yet.

Overview

In this lab you’ll convert an existing users table with 10,000 rows into a distributed table using Citus, splitting it into eight shards across two worker machines. The steps below show the exact psql commands and the expected outputs so you can follow along and verify each stage of the distribution.

Prerequisites

A Citus-enabled PostgreSQL cluster with a control plane and two worker nodes.
A chirp database containing a users table.
psql access from the control plane to the coordinator (example uses -h localhost -p 5432 -U postgres -d chirp).

Step 1 — Confirm the users table row count

First, make sure the users table contains the expected 10,000 rows:

controlplane ~ on ☁️ (us-east-1) → psql -h localhost -p 5432 -U postgres -d chirp -c "SELECT count(*) FROM users;"
 count
-------
 10000
(1 row)

Step 2 — Create the distributed table

Convert the local users table into a distributed table using create_distributed_table. Specify the distribution column (user_id) and the number of shards (shard_count => 8). Note that the named parameter uses =>.

controlplane ~ on ☁️ (us-east-1) → psql -h localhost -p 5432 -U postgres -d chirp -c "SELECT create_distributed_table('users', 'user_id', shard_count => 8);"
NOTICE:  Copying data from local table...
NOTICE:  copying the data has completed

What this does:

Creates 8 shards for the users table.
Distributes the shards across available worker nodes.
Copies existing data from the local table into the distributed shards.

Step 3 — Verify shard placements across workers

Query the Citus distribution metadata to confirm how shards are placed across worker nodes. The pg_dist_shard_placement table shows the node name and how many placements it has:

controlplane ~ on ☁️ (us-east-1) → psql -h localhost -p 5432 -U postgres -d chirp -c "SELECT nodename, count(*) AS placements FROM pg_dist_shard_placement GROUP BY nodename;"
    nodename    | placements
----------------+------------
 worker-1:5432  |          4
 worker-2:5432  |          4
(2 rows)

With 8 shards and 2 worker machines, you should see an even distribution: 4 shard placements per worker.

Quick reference: commands and purpose

Command (example)	Purpose
`SELECT count(*) FROM users;`	Verify local row count before distributing
`SELECT create_distributed_table('users', 'user_id', shard_count => 8);`	Convert `users` to a distributed table with 8 shards
`SELECT nodename, count(*) AS placements FROM pg_dist_shard_placement GROUP BY nodename;`	Confirm shard placement across workers

Notes and troubleshooting

If you see an uneven distribution, ensure both workers are properly registered with the coordinator and that there are no placement constraints. Re-run the placement query after any worker configuration changes.

Why One Database Isn't Enough

Sharding Fundamentals

Routing and Hot Shards

Shard Keys and Partitioning

Resharding and Cross-Shard Queries

Split the Users Table

Overview

Prerequisites

Step 1 — Confirm the users table row count

Step 2 — Create the distributed table

Step 3 — Verify shard placements across workers

Quick reference: commands and purpose

Notes and troubleshooting

Links and references

Watch Video

Practice Lab

​Overview

​Prerequisites

​Step 1 — Confirm the users table row count

​Step 2 — Create the distributed table

​Step 3 — Verify shard placements across workers

​Quick reference: commands and purpose

​Notes and troubleshooting

​Links and references

Watch Video

Practice Lab

Overview

Prerequisites

Step 1 — Confirm the users table row count

Step 2 — Create the distributed table

Step 3 — Verify shard placements across workers

Quick reference: commands and purpose

Notes and troubleshooting

Links and references