Deep Diving the Citus Distribution Models Along with Shard Balancing/Read Scaling

In my previous blog post, Integrating Citus with Patroni: Sharding and High Availability Together, I explored how to integrate Citus with Patroni and demonstrated how basic table distribution works. In this follow-up post, I will discuss various other Citus distribution models. We will also explore how shard rebalancing and data movement are handled and further leverage Read scaling via targeting secondary worker nodes.

Let’s explore different distribution categories one by one. I added the nodes/cluster details below for ease of understanding.

 

 

1) Schema-based sharding

As the name suggests, the different “schema” inside the database will be sharded or segregated over different dedicated worker nodes. We don’t need to distribute any tables, as they automatically reside in their “schema” dedicated shard node.

This can be useful when a multi-tenant application or microservices has a dedicated schema for each service. The application needs to switch [search_path] to use different schemas while working.

 

• • Below, we create two different schemas (s1 and s2) in the coordinator node and then distribute them over the worker nodes.

 

 

 

 

 

 

 

• • We can quickly check some basic information related to an individual schema.

 

 

 

 

• • Here, we can find some more detailed information about each individual table placement. Schema “s1” tables are distributed over the worker node (172.31.95.174), and “s2” tables are distributed over a different worker node (172.31.84.95). Any further new tables will be placed in their designated worker node only.

 

 

 

Query routing:

 

 

• • When we query in schema “s1”, the data is fetched from the Leader worker node (172.31.95.174).

 

 

 

 

• • When we query in schema “s2“, the data is fetched from the Leader worker node (172.31.84.95).

 

 

 

2) Reference tables

When we have some table dependency for the purpose of foreign key relation or read retrievals, we can, with the help of the function “create_reference_table(),” place the reference table on each worker node so that the JOIN query can be run locally. This might improve the retrieval efficiency instead of relying on row retrievals from remote worker nodes. This process basically replicates the underlying parent table on each worker node.

 

• • Here, we have created two tables (r1 and r2) with a FK relationship.

 

 

 

 

 

 

 

• • Then, we are replicating this reference table “r1” across all worker nodes.

 

 

 

 

• • Finally, we distribute the data of the child table “r2” on each Leader worker node.

 

 

 

 

• • Now if we monitor the details we can clearly see the Parent table “r1” is basically replicated to each worker node with the same “shardid” and “colocation_id”.

 

 

 

 

• • However, the Child table “r2” is evenly distributed among all Leader worker nodes.

 

 

 

Query routing:

The output is retrieved from one of the designated leader worker nodes, in accordance with the underlying data distribution.

 

3) Co-location tables

If the distributed tables have the same/common distributed column, then we can co-locate them to enable efficient distributed joins and foreign key relationships. This basically means shards with the same hash range reside in the same node always.

 

• • So, below we create two tables having a common distributed column/FK relationship.

 

 

 

 

• • The child table “r2” is distributed in a manner so that it co-locates with the parent table “r1” based on the common distributed column “id“.

 

 

 

 

• • Then, we performed some DML operations.

 

 

 

 

• • In the below command output, we can see both tables (“r1” and “r2“) shards placed in the same Leader worker node and its replica. The colocation_id is the same for both.

 

 

 

 

 

• • Here is the individual shard distribution in more detail.

 

 

 

 

Query routing:

The query routing rule has not changed much. Like a reference table, we can just perform multiple table-related queries or join operations.

 

Note:- The key difference between standard reference tables and co-location is that, in the former, one of the tables is replicated across all worker nodes, making it locally available. In contrast, with co-location, both tables are distributed across the nodes in such a way that related rows reside on the same worker node.

 

4) Columnar storage

For analytical or data warehousing needs, we can use the columnar format, where the columns are stored on disk rather than rows in a compressed way. This further helps in Write distribution by distributing the data over backend worker nodes.

 

• • Here, we create a table of columnar type and push some random data.

 

 

 

 

 

• • We distributed the table based on the “id” column.

 

 

 

 

Query routing:

The query routing process for columnar type will be the same as the other options we explored earlier.

 

Read scaling via secondary worker nodes

 

 

• • By default, Citus allows performing reads from the worker leader nodes only.

 

 

 

 

• • However, changing the “citus.use_secondary_nodes” from the default “never” to “always” allows the worker node replicas to serve the read-specific requests.

 

 

 

 

• • We can change the read-only setting via direct editing in the Patroni files or dynamically.

 

 

Or

 

 

• • Once the changes are applied, we can see the read requests serving via the Standby nodes.

 

 

 

 

 

 

• • But there is one caveat in the above process as every connection will go to read-only replicas which eventually lead to below errors in case any write activity happens.

 

 

 

 

• • The solution to this problem is changing “citus.use_secondary_nodes” back to “never“, but only inside a particular session/client, as below.

 

 

 

 

Shard balancing/moving

If we want to move a specific shard to a different worker node, we can use the “citus_move_shard_placement()” function.

 

• • Initially, shardid “104756” resides in the Leader Worker node (172.31.95.174).

 

 

 

 

 

• • Then, we moved the shard to another Leader worker node (172.31.84.95).

 

 

 

 

• • Now we can see the shardid “104756” reflected in the different Worker nodes.

 

 

We can also rebalance the shard in case the data distribution is not evenly distributed and one worker has more shards.

 

• • Here, we can see the leader worker node (172.31.84.95) having more shards than (172.31.95.174).

 

 

 

 

• • Now we are rebalancing the shard for table “t1”.

 

 

 

 

• • The table is rebalanced more closely now.

 

 

 

Final thought on Citus Distribution Models

This blog post tries to cover key Citus data distribution methods, detailing their functions and query routing. Another feature, “create_distributed_table_concurrently()”, which allows table distribution without downtime, could be beneficial when dealing with an existing table/data set that needs to be distributed over different worker nodes. The usual data distribution using the function “create_distributed_table()” can block the DML until the table sharding process completes. We also demonstrated how to improve read scalability by routing read queries to secondary worker nodes. Finally, we also touched on the shard rebalancing and movement option, both of which play a crucial role in the data stability and performance.

This blog is purely based on my evaluation of the Citus extension and does not reflect Percona’s recommendations/suggestions. Again, I would emphasize that before using any specific distribution model, please carefully assess its pros and cons and evaluate possible support coverage. Testing in a non-production environment is highly recommended to better understand the model and avoid unforeseen issues in production later on.