Citus with multi-coordinator using Streaming Replication |

In the standard setup of Citus, it has one coordinator, so it is the single point of failure and the bottle neck of the system.

To solve these disadvantages, I built a multi-coordinator using Streaming Replication, which is the built-in replication feature of PostgreSQL.

In this blog, I install and run all postgres servers: two coordinators and two workers, in a single host.

Figure 1:

1. Installing PostgreSQL

Download the latest postgresql source code, and install to /usr/local/pgsql.

# cd /usr/local/src
# tar xvfj postgresql-9.6.3.tar.bz2
# cd postgresql-9.6.3
# ./configure —prefix=/usr/local/pgsql
# make && make install

Add the path /usr/local/pgsql/bin to the environment variable PATH.

# export PATH=/usr/local/pgsql/bin:$PATH

2. Installing Citus

# cd /usr/local/src/postgresql-9.6.3/contrib
# git clone https://github.com/citusdata/citus.git
# cd citus
# configure
# make && make install

3. Creating servers

At first, create one coordinator and two workers.

3.1. Creating the database clusters

Create citus-sr subdirectory, and execute initdb command to create each database cluster.

# cd /usr/local/pgsql
# mkdir citus-sr
# initdb -D citus-sr/coordinator1
# initdb -D citus-sr/worker1
# initdb -D citus-sr/worker2

3.2. Editing postgresql.conf

Edit each postgresql.conf.

coordinator1

listen_addresses = '*'
wal_level = replica
max_wal_senders = 3
hot_standby = on
shared_preload_libraries = 'citus'

worker1

port = 9701
shared_preload_libraries = 'citus'

worker2

port = 9702
shared_preload_libraries = 'citus'

3.3. Starting servers

Start three servers.

# pg_ctl -D citus-sr/coordinator1 -l coordinator1_logfile start
# pg_ctl -D citus-sr/worker1  -l worker1_logfile start
# pg_ctl -D citus-sr/worker2  -l worker2_logfile start

3.4. Creating extension

Execute CREATE EXTENTION command in each server.

# psql -c "CREATE EXTENSION citus;"
# psql -p 9701 -c "CREATE EXTENSION citus;"
# psql -p 9702 -c "CREATE EXTENSION citus;"

3.5. Adding worker nodes

Add workers to the coordinator1.

# psql -c "SELECT * from master_add_node('localhost', 9701);"
# psql -c "SELECT * from master_add_node('localhost', 9702);"

# psql -c "select * from master_get_active_worker_nodes();"
 node_name | node_port 
-----------+-----------
 localhost |      9701
 localhost |      9702
(2 rows)

4. Creating the coordinator’s standby server

Create the coordinator2 server which is a replication of the coordinator1.

At first, create the coordinator2’s database cluster using the pg_basebackup utility.

pg_basebackup -D /usr/local/pgsql/citus-sr/coordinator2 -X stream --progress -U postgres -R

Next, edit the postgresql.conf and recovery.conf files where are located in the citus-sr/coordinator2 subdirectory.

postgresql.conf

port = 5433

recovery.conf

standby_mode = 'on'
primary_conninfo = 'user=postgres port=5432 sslmode=disable sslcompression=1'

Last, start the server.

# pg_ctl -D citus-sr/coordinator2 -l coordinator_logfile start

5. Creating table

Create a table and define the distributed table of the created table on the coordinator1.

# psql
psql (9.6.3)
Type "help" for help.

postgres=# CREATE TABLE sample (id int primary key, data int);
CREATE TABLE
postgres=# SELECT create_distributed_table('sample', 'id');
 create_distributed_table 
--------------------------
 
(1 row)

postgres=# INSERT INTO sample SELECT GENERATE_SERIES(1, 10000), GENERATE_SERIES(1, 10000);
INSERT 0 10000

6. Operating data

If you access to the coordinator1, you can obviously do everything what citus can do.

If you access to the coordinator2, you can execute SELECT commands.

# psql -p 5433
psql (9.6.3)
Type "help" for help.

postgres=# SELECT * FROM sample WHERE id < 8 ORDER BY id;
 id | data 
----+------
  1 |    1
  2 |    2
  3 |    3
  4 |    4
  5 |    5
  6 |    6
  7 |    7
(7 rows)

Interestingly, you can execute DELETE and UPDATE commands using the master_modify_multiple_shards function even if you access to the coordinator2.

postgres=# SELECT master_modify_multiple_shards('UPDATE sample SET data = data*2  WHERE id < 4');
 master_modify_multiple_shards 
-------------------------------
                             3
(1 row)

postgres=# SELECT * FROM sample WHERE id < 8 ORDER BY id;
 id | data 
----+------
  1 |    2
  2 |    4
  3 |    6
  4 |    4
  5 |    5
  6 |    6
  7 |    7
(7 rows)

However, the master_modify_multiple_shards function is not efficient because it executes UPDATE/DELETE commands for all shards of the target table.

Citus with multi-coordinator using Streaming Replication