Manual Query Propagation in Citus Distributed PostgreSQL Clusters

Manual Query Propagation

When a user issues a query, the Citus coordinator splits it into smaller fragments that can run independently on worker shards, allowing the query to be distributed across the cluster. In advanced scenarios you may want to control this propagation manually, and Citus provides utility functions to propagate raw SQL to workers, shards, or placements.

These manual functions bypass the coordinator's logic, locking, and consistency checks, so they should be used with caution to avoid data inconsistency or deadlocks.

Run on All Workers

The coarsest level broadcasts a statement to every worker node, useful for inspecting cluster-wide settings.

-- List the work_mem setting of each worker database
SELECT run_command_on_workers($cmd$ SHOW work_mem; $cmd$);

Note: Do not create database objects on a worker with this command, as it makes automatic addition of worker nodes harder.

Note: The run_command_on_workers function and other manual propagation commands can only execute queries that return a single column and a single row.

Run on All Shards

This level targets every shard of a specific distributed table, allowing access to local metadata such as table statistics.

The run_command_on_shards function interpolates the shard name into the SQL command. The example below estimates the total row count of a distributed table by summing the reltuples from each shard's pg_class table.

-- Get the estimated row count for a distributed table by summing the
-- estimated counts of rows for each shard.
SELECT sum(result::bigint) AS estimated_count
FROM run_command_on_shards(
  'my_distributed_table',
  $cmd$
    SELECT reltuples
    FROM pg_class c
    JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
    WHERE (n.nspname || '.' || relname)::regclass = '%s'::regclass
      AND n.nspname NOT IN ('citus', 'pg_toast', 'pg_catalog')
  $cmd$
);

Run on All Placements

The finest granularity runs a command on every placement (shard replica). This is useful for data‑modifying statements that must affect each replica to maintain consistency.

Example: updating an updated_at column on every row of a distributed table by issuing an UPDATE on each placement.

-- note we're using a hard‑coded date rather than a function such as "now()"
-- because the query will run at slightly different times on each replica
SELECT run_command_on_placements(
  'my_distributed_table',
  $cmd$
    UPDATE %s SET updated_at = '2017-01-01';
  $cmd$
);

The companion function run_command_on_colocated_placements inserts the names of two colocated placements into the query, ensuring the same worker is chosen for both.

-- Suppose we have two distributed tables
CREATE TABLE little_vals (key int, val int);
CREATE TABLE big_vals    (key int, val int);
SELECT create_distributed_table('little_vals', 'key');
SELECT create_distributed_table('big_vals',    'key');

-- Trigger function to double the value when a row is inserted into little_vals
CREATE OR REPLACE FUNCTION embiggen() RETURNS TRIGGER AS $$
BEGIN
  IF (TG_OP = 'INSERT') THEN
    EXECUTE format('INSERT INTO %s (key, val) SELECT ($1).key, ($1).val*2;', TG_ARGV[0]) USING NEW;
  END IF;
  RETURN NULL;
END;
$$ LANGUAGE plpgsql;

-- Relate the colocated tables with a trigger on each colocated placement
SELECT run_command_on_colocated_placements(
  'little_vals',
  'big_vals',
  $cmd$
    CREATE TRIGGER after_insert AFTER INSERT ON %s
      FOR EACH ROW EXECUTE PROCEDURE embiggen(%L)
  $cmd$
);

Limitations

Multi‑statement transactions have no built‑in deadlock safety.

There are no safeguards for intermediate query failures that could cause inconsistency.

The functions cache results in memory and cannot handle very large result sets.

If a node cannot be reached, the functions fail early.

Improper use can lead to serious data corruption.