Citus Distributed PostgreSQL: Inserting, Updating, and Deleting Data (SQL Reference)

Inserting data

Use the regular INSERT command to add rows to a distributed table. The distribution column determines the target shard, and the query is routed to that shard for execution.

INSERT INTO github_events VALUES (2489373118,'PublicEvent','t',24509048,'{}','{"id": 24509048, "url": "https://api.github.com/repos/SabinaS/csee6868", "name": "SabinaS/csee6868"}','{"id": 2955009, "url": "https://api.github.com/users/SabinaS", "login": "SabinaS", "avatar_url": "https://avatars.githubusercontent.com/u/2955009?", "gravatar_id": ""}',NULL,'2015-01-01 00:09:13');
INSERT INTO github_events VALUES (2489368389,'WatchEvent','t',28229924,'{"action": "started"}','{"id": 28229924, "url": "https://api.github.com/repos/inf0rmer/blanket", "name": "inf0rmer/blanket"}','{"id": 1405427, "url": "https://api.github.com/users/tategakibunko", "login": "tategakibunko", "avatar_url": "https://avatars.githubusercontent.com/u/1405427?", "gravatar_id": ""}',NULL,'2015-01-01 00:00:24');

Multiple INSERT statements can be combined into a single multi‑row INSERT for efficiency:

INSERT INTO github_events VALUES
  (2489373118,'PublicEvent','t',24509048,'{}','{"id": 24509048, ... }', ... ,NULL,'2015-01-01 00:09:13'),
  (2489368389,'WatchEvent','t',28229924,'{"action": "started"}', ... ,NULL,'2015-01-01 00:00:24');

"From Select" clause (distributed roll‑ups)

Citus supports INSERT ... SELECT, allowing rows to be populated from a query result. When both source and target tables share the same distribution column, the statement is pushed down and executed in parallel on all workers. Certain SQL features (e.g., ORDER BY, LIMIT, GROUP BY on non‑distribution keys, window functions, and cross‑shard joins) prevent this optimization.

If the source and target are on different nodes and repartitioning cannot be applied, Citus falls back to a three‑step method that routes results to the coordinator and then redistributes them, which is less efficient.

Use EXPLAIN to see which path Citus chooses, and consider disabling repartitioned inserts via citus.enable_repartitioned_insert_select for very large shard counts.

COPY command (bulk loading)

Download the example GitHub archive CSV files and load them with \COPY:

wget http://examples.citusdata.com/github_archive/github_events-2015-01-01-{0..5}.csv.gz
gzip -d github_events-2015-01-01-*.gz
\COPY github_events FROM 'github_events-2015-01-01-0.csv' WITH (format CSV);

Note: COPY does not provide cross‑shard snapshot isolation. Concurrent COPY and SELECT may succeed on some shards and not on others, potentially leaving small gaps in recent data.

Using roll‑up cache aggregation

For high‑frequency queries, pre‑compute aggregates into a summary table. Example schema for page‑view tracking:

CREATE TABLE page_views (
  site_id int,
  url text,
  host_ip inet,
  view_time timestamp default now(),
  PRIMARY KEY (site_id, url)
);
SELECT create_distributed_table('page_views','site_id');

Aggregate daily page views:

SELECT view_time::date AS day, site_id, url, count(*) AS view_count
FROM page_views
WHERE site_id = 5
  AND view_time >= DATE '2016-01-01'
  AND view_time < DATE '2017-01-01'
GROUP BY view_time::date, site_id, url;

Store the result in a daily_page_views table to avoid recomputation and reduce storage growth:

CREATE TABLE daily_page_views (
  site_id int,
  day date,
  url text,
  view_count bigint,
  PRIMARY KEY (site_id, day, url)
);
SELECT create_distributed_table('daily_page_views','site_id');
INSERT INTO daily_page_views (day, site_id, url, view_count)
SELECT view_time::date, site_id, url, count(*)
FROM page_views
WHERE view_time >= DATE '2017-01-01' AND view_time < DATE '2017-01-02'
GROUP BY view_time::date, site_id, url;
SELECT day, site_id, url, view_count FROM daily_page_views WHERE site_id = 5 AND day >= DATE '2016-01-01' AND day < DATE '2017-01-01';

When new rows arrive for an existing day, use ON CONFLICT upserts to increment the count:

INSERT INTO daily_page_views (day, site_id, url, view_count)
SELECT view_time::date, site_id, url, count(*)
FROM page_views
WHERE view_time >= DATE '2017-01-01'
GROUP BY view_time::date, site_id, url
ON CONFLICT (day, url, site_id) DO UPDATE
  SET view_count = daily_page_views.view_count + EXCLUDED.view_count;

Update and Delete Standard UPDATE and DELETE work on distributed tables. Example:

DELETE FROM github_events WHERE repo_id IN (24509048,24509049);
UPDATE github_events SET event_public = TRUE WHERE (org->>'id')::int = 5430905;

When a statement touches multiple shards, Citus uses a single‑phase commit by default. Enable two‑phase commit for safety with SET citus.multi_shard_commit_protocol = '2pc'; . Single‑shard updates run on the worker node without 2PC. For row‑level locking, Citus supports SELECT ... FOR UPDATE on hash‑distributed and reference tables with replication factor 1. Example transaction:

BEGIN;
SELECT * FROM github_events WHERE repo_id = 206084 FOR UPDATE;
-- calculate new value in application code
UPDATE github_events SET event_public = :new_value WHERE repo_id = 206084;
COMMIT;

Maximizing write performance On large machines, INSERT and UPDATE/DELETE can scale to roughly 50,000 queries per second when using many parallel, long‑lived connections and proper lock handling. See the “Scaling Data Ingestion” section of the Citus documentation for details. For more information, refer to the Citus documentation links embedded throughout the guide.