Quick Guide: Testing Citus Distributed PostgreSQL on Kubernetes with Distributed, Co‑located, Reference, and Columnar Tables

Preparation

Assume a Citus‑extended PostgreSQL cluster is already deployed on k8s. Verify the cluster with kubectl get po -n citus, which shows one coordinator pod and three worker pods.

NAME                  READY   STATUS    RESTARTS   AGE
citus-coordinator-0   2/2     Running   0          3h55m
citus-worker-0        2/2     Running   0          22m
citus-worker-1        2/2     Running   0          21m
citus-worker-2        2/2     Running   0          21m

Enter the coordinator container and list active workers:

psql 'host=citus-coordinator user=postgres' -c "SELECT * FROM citus_get_active_worker_nodes();"

node_name                      | node_port 
-----------------------------------------------------+-----------
citus-worker-1.citus-worker.citus.svc.cluster.local |      6432
citus-worker-2.citus-worker.citus.svc.cluster.local |      6432
citus-worker-0.citus-worker.citus.svc.cluster.local |      6432
(3 rows)

Creating a Distributed Table

Define the events table and distribute it on device_id:

CREATE TABLE events (
  device_id bigint,
  event_id bigserial,
  event_time timestamptz default now(),
  data jsonb not null,
  PRIMARY KEY (device_id, event_id)
);

SELECT create_distributed_table('events', 'device_id');

Insert one million random events to demonstrate automatic routing:

INSERT INTO events (device_id, data)
SELECT s % 100, ('{"measurement":' || random() || '}')::jsonb
FROM generate_series(1,1000000) s;

Fetch the last three events for device_id = 1; the query runs on a single worker and returns in 4.8 ms.

SELECT * FROM events WHERE device_id = 1 ORDER BY event_time DESC, event_id DESC LIMIT 3;

device_id | event_id |          event_time           |                data                 
-----------+----------+-------------------------------+-------------------------------------
         1 |   999901 | 2022-03-24 02:30:50.205478+00 | {"measurement": 0.8822990134507691}
         1 |   999801 | 2022-03-24 02:30:50.205478+00 | {"measurement": 0.5239176115816448}
         1 |   999701 | 2022-03-24 02:30:50.205478+00 | {"measurement": 0.9900647926398349}
(3 rows)
Time: 4.779 ms

Explain the plan for SELECT count(*) FROM events. The plan shows a Custom Scan (Citus Adaptive) with 32 parallel tasks; execution time 5.4 ms.

QUERY PLAN
-----------------------------------------------
 Aggregate  (cost=250.00..250.02 rows=1 width=8)
   Output: COALESCE((pg_catalog.sum(remote_scan.count))::bigint, '0'::bigint)
   -> Custom Scan (Citus Adaptive)  (cost=0.00..0.00 rows=100000 width=8)
        Task Count: 32
        ...
Time: 5.427 ms

Using Co‑location to Create Distributed Tables

Create a devices table, index the device_type_id column, and distribute it with explicit co‑location to the events table:

CREATE TABLE devices (
  device_id bigint primary key,
  device_name text,
  device_type_id int
);
CREATE INDEX ON devices (device_type_id);
SELECT create_distributed_table('devices', 'device_id', colocate_with := 'events');

Insert device metadata:

INSERT INTO devices (device_id, device_name, device_type_id)
SELECT s, 'device-'||s, 55 FROM generate_series(0,99) s;

Add a foreign‑key constraint to enforce referential integrity:

ALTER TABLE events ADD CONSTRAINT device_id_fk FOREIGN KEY (device_id) REFERENCES devices (device_id);

Compute the average measurement for devices of type 55. The query runs in parallel across shards and finishes in 122.5 ms.

SELECT avg((data->>'measurement')::double precision)
FROM events JOIN devices USING (device_id)
WHERE device_type_id = 55;

avg         
--------------------
 0.4997412230952178
(1 row)
Time: 122.548 ms

Creating a Reference Table

Create a device_types table and replicate it to all nodes with create_reference_table:

CREATE TABLE device_types (
  device_type_id int primary key,
  device_type_name text not null unique
);
SELECT create_reference_table('device_types');
INSERT INTO device_types (device_type_id, device_type_name) VALUES (55, 'laptop');
ALTER TABLE devices ADD CONSTRAINT device_type_fk FOREIGN KEY (device_type_id) REFERENCES device_types (device_type_id);

Join the three tables to fetch the last three events for devices whose type name starts with “laptop”. The query completes in 96.5 ms.

SELECT device_id, event_time, data->>'measurement' AS value, device_name, device_type_name
FROM events JOIN devices USING (device_id) JOIN device_types USING (device_type_id)
WHERE device_type_name LIKE 'laptop%'
ORDER BY event_time DESC LIMIT 3;

device_id |          event_time           |        value        | device_name | device_type_name 
-----------+-------------------------------+---------------------+-------------+------------------
        31 | 2022-03-24 02:30:50.205478+00 | 0.9994211581289107  | device-31   | laptop
        31 | 2022-03-24 02:30:50.205478+00 | 0.13771543211483106 | device-31   | laptop
        88 | 2022-03-24 02:30:50.205478+00 | 0.5585740912470349  | device-88   | laptop
(3 rows)
Time: 96.537 ms

Creating a Columnar Table

Create a columnar table for events:

CREATE TABLE events_columnar (
  device_id bigint,
  event_id bigserial,
  event_time timestamptz default now(),
  data jsonb not null
) USING columnar;

Bulk‑load ten million rows:

INSERT INTO events_columnar (device_id, data)
SELECT d, '{"hello":"columnar"}' FROM generate_series(1,10000000) d;

Create a row‑based copy for size comparison:

CREATE TABLE events_row AS SELECT * FROM events_columnar;

Inspect table sizes. The columnar table occupies 25 MB, while the row‑based copy occupies 806 MB—a compression factor of dozens.

\d+
List of relations
 Schema |            Name            |   Type   |  Owner   | Persistence | Access method |   Size   | Description 
--------+----------------------------+----------+----------+-------------+---------------+----------+-------------
 public | events_columnar            | table    | postgres | permanent   | columnar      | 25 MB    | 
 public | events_row                 | table    | postgres | permanent   | heap          | 806 MB   | 
(2 rows)

Columnar tables support bulk loading via COPY or INSERT..SELECT for optimal compression but do not support UPDATE, DELETE, or foreign‑key constraints. They can be combined with partitioned tables that use row storage for newer partitions.