How ACID is Citus? (compared to YugabyteDB)
Tags: distributed, sql, database, yugabytedb
Published: 2023-04-26 09:57:00

What is the difference between **sharding over monolithic databases**, like the CitusDB extension for PostgreSQL, and **Distributed SQL**? Both are distributing the table rows and index entries but distributing **SQL** means more than that: distributing **SQL transactions** and keeping their **ACID** properties.
- **Sharding monolithic databases** will use local transactions, with a coordinator above it that uses two-phase commit to commit on all shards and detect if there are in-doubt transactions
- **Distributed SQL databases** run global transactions, so that applications can see it as one logical database, to provide horizontal scalability to OLTP applications.

The most obvious difference is with ACID properties. Transactions must be **atomic** (A), moving the database from one state to the other, and **isolated** (I), so that other users see either the state before or the state after.

A typical example is a bank account transfer. If you transfer 100$ from Account A to Account B, the database transaction doing the transfer updates the two accounts one after the other to add 100$ to B an withdraw 100$ to B. The total amount of all accounts may temporarily be off by 100$ when queried by the transaction doing the transfer, but all other transactions should never see this intermediate state.

# pgbench account transfer

To give an example, I'll use `pgbench` on YugabyteDB. I simply initialize the default table, which includes the `pgbench_accounts` table:
```sh
pgbench -i
```
This creates 100000 accounts with a balance of zero:
```sql
select count(*),min(abalance),max(abalance),sum(abalance) 
 from pgbench_accounts;
```
I create a custom script to do a simple transfer between two random accounts:
```sql
cat > /tmp/pgbench-acid.sql <<'SQL'
\set aid1 random(1, 100000 * :scale)
\set aid2 random(1, 100000 * :scale)
\set delta random(-5000, 5000)
BEGIN;
UPDATE pgbench_accounts SET abalance = abalance + :delta WHERE aid = :aid1;
UPDATE pgbench_accounts SET abalance = abalance - :delta WHERE aid = :aid2;
END;
SQL
```

## YugabyteDB Distributed SQL

I run this in the background for 5 minutes from 10 connections:
```sh
pgbench -f /tmp/pgbench-acid.sql -T 300 -c 10 &
```
Now from `psql` I query the total account balance every 100 milliseconds:
```sql
select count(*),min(abalance),max(abalance),sum(abalance) 
 from pgbench_accounts;
\watch 0.1
```

Thanks to all transactions being ACID, the total amount never changes because all transfers are from one account to another:
![YugabyteDB where the total amount is always at zero](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/d1ven5lrggfxetb8ri7r.gif)
Those are distributed transactions, with reads and writes on all nodes:
![rad and write operations on all YugabyteDB nodes](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/qhoyrqm6zajautovi9w3.png)

## PostgreSQL with CitusDB

Let's run the same on CitusDB. I start a 3 nodes cluster:

```sh
git clone https://github.com/citusdata/docker.git citusdata-docker
cd citusdata-docker
docker-compose up -d --scale worker=3 
```
I initialize `pgbench` tables:
```sh
pgbench -i
```
I distribute the `pgbench_accounts` table:
```sql
docker exec -it citusdata-docker_master psql -c '
 select create_distributed_table('pgbench_accounts', 'aid');
'
```
I run the same script in the background:
```sh
pgbench -f /tmp/pgbench-acid.sql -T 300 -c 10 &
```
and query the total amount every 100 milliseconds:
```sql
select count(*),min(abalance),max(abalance),sum(abalance) 
 from pgbench_accounts;
\watch 0.1
```
The total amount is sometimes negative and sometimes positive because ongoing transactions are transferring random amounts. The distributed query see some intermediate states.
![CitusDB showing amounts different than zero](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/oqvh08t4uup6mb39g0dc.gif)

This is not ACID, it is Eventual Consistency. At some point, when there are no ongoing transactions anymore, the shards will be synchronized.

Note that this example was only about the **A**tomicity and **I**solation of ACID. The **C**onsistency is about integrity constraints and CitusDB do not allow cross-shard foreign keys or even global indexes. That is another difference with Distributed SQL (see this previous [post](https://dev.to/yugabyte/is-cosmosdb-a-new-sql-database-is-citusdb-a-distributed-sql-did-hyperscale-vanished-in-the-hyperspace-472d) about it). The **D**urability is about persistence. This CitusDB is not protected. You should add multiple standby databases, with synchronous commit, for each shard to protect it. This increases the complexity for more than a handful of shards (Patroni templates can help to get this complexity correctly designed). The YugabyteDB cluster above has replication built-in alongside the distribution and was a Replication Factor 3 cluster that can scale to more nodes transparently because the tables are actually distributed to smaller shards that can move online.

## CitusDB use cases

The behavior of **eventually consistent reads** for multi-shard queries is expected and is acceptable for the CitusDB use cases that it is designed for (see [When to Use Citus](https://docs.citusdata.com/en/v10.2/get_started/what_is_citus.html#when-to-use-citus)):
- Real-Time Analytics: The tables are usually sharded on time and data ingest goes to one shard. Then, the queries will be consistent.
- Multi-Tenant Application: For a few pre-defined tenants, the tables are sharded per tenant and transactions will be within one tenant, and will be consistent.

## YugabyteDB use cases

**Distributed SQL** databases (Google Spanner, CockroachDB, TiDB, YugabyteDB) guarantee ACID properties for cross-shard transactions and are then consistent for **all OLTP applications** which involve multi-shard transactions, and allow automatic re-sharding when scaling-out.

Among them, the only one which is **PostgreSQL-compatible** when it comes to transactions is YugabyteDB as the others do not have the same features and behavior. YugabyteDB supports all PostgreSQL isolation levels, with the same behavior, for all distributed transactions.