Data locality: Scaling PostgreSQL with Citus intelligently

09.2023 / Category: , / Tags: | |

While sharding is often advertised as “THE solution to PostgreSQL scalability”, it is necessary to keep some technical aspects in consideration in terms of performance. The rule is: Sharding should not be used without a deeper awareness of what it is you are actually doing to the data. It's important to keep in mind that sharding has to be applied in a clever and thoughtful manner. One of the most common mistakes is to ignore the concept of “data locality”. It's important for many IT problems, but crucial in the context of database sharding. Citus is one of the most sophisticated sharding solutions in the PostgreSQL world. It can help you to achieve maximum scalability and allows for efficient analytics as well as OLTP. Citus is available on-premise or as part of the Microsoft Azure cloud.

What is data locality? Let's take a look together.

Preparing data for sharding

To demonstrate the concept, we first have to create two tables. For the sake of simplicity, we'll use customers and sales:

The data model is really straightforward. In this scenario, the typical way to analyse the data is to join the customer with the sales table. Why is this relevant?

To understand it, first let's distribute the table and add some data:

Note that the data is sharded using the “id” which is not the join criteria.

In the next step, we can load some data:

Running queries across PostgreSQL shards

Once the data has been loaded, it's time to execute a query. All we want to do is to join and count some data. Here's what happens:

Oops! Citus refuses to run the query because the join doesn't use the sharding criteria in the data model.

Citus will send an error message by default and refuse to run the query. Why is that the case? Imagine we want to join two rows on the same machine: PostgreSQL will do some magic in RAM locally and voila, two rows has been join (it is not as simple as you can see in our blog post about PostgreSQL join strategies but one can understand the logic). Local joins are efficient but if all the data has to go through an expensive network stack? Things will be orders of magnitude more expensive and in case of large data sets this strategy becomes totally impractical and doomed to fail. Therefore it is better to error out than to lead the user directly into a serious performance problem.

In some cases, it might still be necessary to force a join. However, this requires a configuration variable which can be set locally. Note that this is not something you should do in real life on large data sets:

Due to the small data set we are using here (for the sake of simplicity), the query does indeed return in reasonable time. But as stated before, if you do this on large data sets which are not on the same machine as my little test, this query might not end at all.

To understand what's going on, we can take a look at the execution plan of the query:

We need around 1.3 seconds to run the query. If the shards reside on different machines, and not just on different ports, this execution time can easily increase by many times.

Joining sharded data with Citus properly

To fix the problem, we have to introduce the concept of “data locality”. The importance of this idea cannot be underestimated. The core idea is that data which has to be joined should stay in the same shard. Do not move data between shards and avoid large operations across shards as much as you can.

In my example, what we have to do is to structure our tables in a way that the same sharding criteria is used for BOTH tables:

Fortunately Citus is nice enough to warn us that we have just created a local table with all the data. The create_distributed_table function DOES NOT delete local data for safety reasons so we have to truncate it locally to avoid storing it twice:

Storing data locally is usually a bad idea in any case, since you can easily run out of space. Sharding and Citus is all about scalability, so moving data to a central place might not be the best of ideas, due to space constraints and other limitations which are diametrically opposed to the core ideas of sharding and scalability.

Once the data has been properly sharded, we can run the query again and see what happens:

Wow! The query is 15 times faster than before.

In reality, the performance difference is often even larger, but still: This is important and should be taken seriously.

Finally …

If you're new to sharding and Citus for PostgreSQL, you might want to check out some of our other blog posts related to scalability.


In order to receive regular updates on important changes in PostgreSQL, subscribe to our newsletter, or follow us on Twitter, Facebook, or LinkedIn.

0 0 votes
Article Rating
Notify of
Inline Feedbacks
View all comments
CYBERTEC Logo white
CYBERTEC PostgreSQL International GmbH
Römerstraße 19
2752 Wöllersdorf

+43 (0) 2622 93022-0

Get the newest PostgreSQL Info & Tools

    This site is protected by reCAPTCHA and the Google Privacy Policy & Terms of Service apply.

    CYBERTEC PostgreSQL International GmbH
    Would love your thoughts, please comment.x
    linkedin facebook pinterest youtube rss twitter instagram facebook-blank rss-blank linkedin-blank pinterest youtube twitter instagram