Database Sharding

First question : Why is it required?

Let’s say you have a single node ( = server) as a database, which is serving your traffic comfortably. Once this traffic increases, you will need to scale. This can include vertical scaling (beefing up the server itself), or horizontal scaling (adding more nodes into the database cluster).

Vertical scaling has limits - you can’t go beyond a super computer. Therefore, in the end you’ll need to consider horizontal scaling. The way this is handled is called sharding.

Avoiding sharding?

Sharding is a complex topic, and even more complex to implement properly. Before deciding to adding this functionality to your database, you should exhaust all other options. We will discuss on other options first.

Partitioning within the single node

The data inside the single node is increasing - leading to an increase in the read and write times. In order to mitigate this, without a drastic measure, we can partition the table itself.

We can divide the primary key into ranges, and each such range is given a partition. Because of this, the indexes built over this data (like B-tree), become smaller, leading to decreased read and write times.

Note that all these partitions still reside inside the same node, so physically no change has been done. The table is logically partitioned, but it still resides within the same node.

Also note that we still would need to maintain a metadata store, which knows which ranges are held by which partition, so that it can route the write & read requests properly.

Replication

Let’s say partitioning didn’t work, and you have reads & writes being slowed down. One more thing for read heavy loads can be done is - replication. We can create a master-replica architecture, where writes happen to a master node, but reads happen from the replicas. Note that the whole table is replicated, not just some parts of it.

Whenever there is a write in the master node, it broadcasts this information to all the replicas, which then update their copy of the table.

Problem #1 : Write load not decreased

Since writes are still happening to a single master node, we still have the issue of writes not scaling up - they are taking more time than expected.

One thing that can be done here is vertically scaling the master node - so that it can handle the increased writes.

Problem #2 : Reads are eventually consistent

Once a write is done on the master node, it still needs to send the notification to the replicas, which then need to write this to their copy. This takes time, and therefore if you want to read just after writing, you might not get the latest value that you wrote, as that might not have been propagated into the replica yet.

Sharding : How can it help?

By dividing the database table itself on multiple nodes, we allow infinite horizontal scaling, theoretically. The writes can be routed to the specific node in which the partition is present, which has the data related to the key. Again, reads can also be routed in a similar fashion.

This solves the issues arising from the need to scale for a higher load.

Sharding

Which data resides on which node?

We have multiple nodes ( = servers) that can store data for the same database table. As part of sharding, we have decided to put different partitions in different nodes. Note that this doesn’t imply that 1 node will only contain only a single partition. It implies that a single partition will reside only inside a single node (and possibly it’s replicas)

How should this partitioning be done? We will need to decide in which node should an item reside if it were to be written.

Partitioning by key range

We can decide to split the primary key range into sub-ranges, which can denote individual nodes where this data can reside. For example, let’s say we are saving the dictionary inside our table, we can distribute words starting with a-q in a single partition, and r-z in another partition.

This means that if I want to read / write “car” then I will be dealing with the first partition, and if I want to read / write “rose” then I will be dealing with the second partition.

An obvious flaw with this partitioning scheme is distribution of words in both the partitions. The r-z partition will have more words, therefore this partition is skewed.

Therefore it is critical to specify the boundaries of a partition properly - data base administrators typically decide the partitioning boundaries.

In some cases, it is possible that a single partition is very hot because of the access pattern. For example, if we are saving timeseries data, then because of key range based partitioning, we will be writing to the same partition mostly.

Partitioning by hash of the key

We can use hash functions which guarantee uniform distribution even for keys that are close. The output of the hash can be used to decide to which partition the item is written / read from.

This avoids the issue of deciding where the partition boundaries lie & hot partitions because of close key writes.

Because of random data being stored inside a partition, we lose the ability to do efficient range queries, since the items are no longer present inside the table in a range format.

How to create secondary indexes now?

Local indexes / document-partitioned indexes

A local index is specific to a partition. You need to specify the partition key AND the index value while making a read request.

This is implemented by keeping track of writes to specific indices inside the partition itself. When a read on the index is done, this information is queried to get the index specific information.

Global indexes / term-partitioned indexes

A global index is not specific to partitions. Rather it encompasses the whole table.

Index terms are distributed amongst partitions such that they store information related to all the items for that term within the whole table. This “distribution” again, is a partitioning problem - in which node should the index “term’s” information be saved?

We can again use either a key range, or a key hash based approach. In the first one, we will compromise randomness (and therefore might encounter hot partitions). In the second one we will compromise efficient range queries.

Rebalancing partitions

Why?

The horizontal scaling enabled by “sharding” - means that we can add more nodes to the system, and the database is able to scale gracefully because of this node’s addition. This process will involve “re-balancing” existing partitions, so that the old nodes shift data to this newly added node.

There are three approaches to rebalancing partitions.

Approach #1 : Have a fixed number of partitions >> number of nodes

Under this approach, we will have a high number of partitions from the get to. So we can have initially 1000 partitions, and 10 nodes. Each node will then have 100 partitions. If we add another node to the system, then we will have 1000/11≈91 partitions per node. This would mean that each existing node will shift around 9 partitions to the new node.

Note that the partition → key mapping doesn’t change in this case. The node → partition mapping, on the other hand, does change. So change will be reflected inside the metadata store, which stores information on which partition is present inside which node.

Note that this “movement” of partition from an older node to a newer node is a costly operation, because all the data will have to be moved over the network.

Pros	Cons
Simple	Deciding the initial number of partitions is not simple
	Initially the partition will have a very low amount of data, and later partitions will have a high amount of data, as more data is ingested by the database. We can not go ahead and split / merge existing partitions on the basis of data size

Approach #2 : Dynamic partitioning

This is simply based on thresholds for the size of a partition. If the size crosses a threshold, then the partition is split into smaller partitions, and moved to a different node. This would update the partition → key mapping. Similarly if there are multiple partitions smaller than a certain threshold, then they are merged together to create a larger partition.

Pros	Cons
Number of partitions grow / shrink according to data size - so partitions don't grow to abnormal sizes, nor there are multiple partitions with approximately no data	Complex
	Initially a single partition will suffer all reads & writes - can be mitigated by setting initial partitions

Approach #3 : Partitions proportional to the number of nodes

In this approach, each node has a fixed number of partitions. We can increase partitions, by simply adding a new node in the system.

This new node, will get data from randomly selected partitions (from existing nodes), which will be split to send the data to the new node. The old partitions will be split in half and one half of the data is sent to the new partition inside the new node, and the other half stays with the old partition inside the old node.

Now in the case of key hash partitioning, this “splitting” activity is simple. We go ahead and split the partition from the middle. But in the case of key range partitioning, the “middle” split might not work because of the data skew issue mentioned above. The first half of the range might have no data, and the other half of the range might have all the data. This makes the splitting of the partition hard in the case of key range partitioning, and easy in the case of key hash partitioning.

Request routing

When a request comes to a sharded database, we get the partition information from the key - either via key range partitioning or key hash partitioning. Now since there are multiple nodes in this system, we need to understand which particular node has this partition.

This information can be stored at three places:

• Store partition ↔ node mapping inside the client library
• Store partition ↔ node mapping inside a custom routing tier
• Store partition ↔ node mapping inside each node

Now consider this case:

• Partition P1 was present inside node N1 - this information is present inside the partition ↔ node mapping wherever that is present
• We are using a “fixed” number of partitions, which are transferred if new nodes are added
• We then reach an upper threshold in the total amount of data present, therefore we add a new node N2, and shift P1 to N2

Now if the partition ↔ node mapping is not updated correctly and in time, the request will actually route to N1 instead of N2.

The handling of this, and more complicated scenarios is done by a custom coordination service like Zookeeper, which tracks cluster metadata. Whenever there are updates in the partition, the nodes notify ZooKeeper, and it notifies the routing tier (wherever that maybe present), to update the routing information.

An assignment looks like this:

What’s the difference between “routing tier” and “ZooKeeper”?

ZooKeeper is responsible for the actual “storage” & “updation” of the routing information for the sharded database system. The “action” of routing is actually handled by the routing tier, using the information present inside ZooKeeper.

Instead of the fancy ZooKeeper, can’t we just use a simple memory based alternative?

Let’s say instead of ZooKeeper we use another “config” server and it’s memory for saving the routing information, that looks like the diagram mentioned above.

The first issue is - reliability - this server can go down, and then the whole database system will go down, because routing doesn’t work anymore. This will make this single “config” server the single point of failure for this system.

Now if we fix reliability by keeping replicas of this simple config server, and keep a master - replica architecture, where writes happen in the master and the reads happen from the replicas - we have the issue of consistency.

This means that even if some write has happened, to update the partition to node mapping, the read right after this write might not get this updated value, as the change has not yet propagated to the replica. ZooKeeper also solves this issue.

Appendix

Sources

• When should you shard your database? - YouTube
• What is DATABASE SHARDING? : https://youtu.be/5faMjKuB9bc?si=F_pRqYbe1Xed_h3i
• What is Database Sharding? : https://youtu.be/hdxdhCpgYo8?si=KaFH2X_9GeC33L6z
• Sharding with Amazon Relational Database Service : Sharding with Amazon Relational Database Service | AWS Database Blog
• Best Practices Working with Billion-row Tables in Databases : - YouTube
• Sharding - MongoDB Manual v7.0
• Choose a Shard Key - MongoDB Manual v7.0
• Designing data intensive applications - Chapter 6

Log structured merge trees with compaction strategies

 

Log structured merge trees

Problem statement & requirements

Where is it used?

A data base is just a server. We need to implement the specific software that is going to work as a database. Since this is going to be a server anyhow, it will have the same components : a CPU, a memory and a disc. Using these, we need to cater to our specific use case to create a data base software.

For LSMs, we serve the use case which is very write heavy. A (losing) competitor of LSM, for this specific use case is a B tree / B + tree. But a B/B+ tree’s write is $O(log_{m}​N)$. This means to write an item to B/B+ trees, you might need to update the disc $O(log_{m}​N)$ times, which might seem small, but given the huge amount of data present inside these databases - this will also take time.

Therefore for write-heavy use cases, we need a different solution. This is where LSMs come into picture.

Why specifically LSM?

LSM offer a write complexity of O(1), and a read complexity of $O(N\times logM)$. (Where $N$ is the number of SSTables scanned, and $M$ is largest SSTable’s size)

Notice that in a B / B + tree, $O(log_{m}​N)$ is the read complexity. So we are compromising on the read complexity while using LSMs.

Design : How does it work?

Writes

All the writes to a LSM based database happen inside the memTable inside the memory. MemTable is a sorted tree / map, which keeps the entries in a sorted fashion whilst the insertion process is working. Now, when this memTable is full, this memTable is flushed into the disc as a SSTable (Sorted String Table). All further writes will happen in a new memTable.

Reads

During a read, we firstly check whether the key is present inside the memTable. If it is, then the corresponding value is returned. Note that we don’t need to check the disc in this case, because whichever value for that key is present in the disc, is older than the one present inside the memTable, as the memTable is created later.

If the item is not present inside the memTable, we start scanning SSTables inside the disc. Since these are already sorted, we take $O(logM)$ time to search these SSTables, where M is the size of the SSTable. Now we start searching for the key from the highest level L0. If they key is found, the corresponding value is returned, without searching for the key in lower levels.

In the worst case scenario, we will need to search all SSTables, inside all the levels. So if there are S SSTables, then we will take $O(S\times log_{2}​M)$ time to search an item.

Bloom filters - an optimisation for reading SSTables

A pre-cursor for searching the SSTable directly is using a bloom filter. A bloom filter can give out the detail on whether the item is definitely not present inside the SSTable. Otherwise it can give that the item might be present in the SSTable.

So during a read - we can firstly check the bloom filter, and see whether the item is definitely not present. If item is definitely not present, there is no need to search this SSTable. Otherwise if the bloom filter says that the item might be present, then we will need to search the SSTable in that case.

Therefore, when a memTable is written to disc, the corresponding bloom filter is also created and attached to the SSTable.

Note that the read complexity still remains the same in the worst case scenario.

How do bloom filters work?

Bloom filters are just hash sets. If some item is input inside the bloom filter, then it’s key is hashed and the output bit is marked as 1. Now the range of the output of the hash function is small, therefore many keys will map to a single output bit.

Therefore, if for a key, the output bit of the hash function is present inside the bloom filter hash set, then the key might be present inside the corresponding data structure.

But if for a key, the output bit of the hash function is NOT present inside the bloom filter hash set, then the key definitely is not present inside the corresponding data structure - the SSTable in this case.

Compaction - the background process to optimise reads

As certain conditions are met, a level merges it’s SSTables and flushes the output in a lower level. This is a background process in LSM trees, known as compaction. There are different strategies to implement this compaction mechanism.

Fundamental concepts

Size amplification / Space amplification

The actual size of the table and the optimal size of the table can differ, because multiple versions of the same key are present inside different levels of a LSM tree.

This increase in actual size is called space amplification

Write amplification

This refers to the total number of times a disc write can happen because of a single write operation on the LSM tree.

In the case of LSMs, one write operation can lead to writes into memTable, then L0 SStable, L1 SStable etc. Therefore a single write can lead to many disc writes.

Read amplification

Similar to write amplification, this refers to the total number of times a disc read can happen because of a single read operation on the LSM tree.

In the case of LSMs, one read operation will need to read through memTable and multiple SS tables through multiple levels inside disc.

Strategy 1 : Size tiered compaction

In size tiered compaction, a background process groups SS tables of similar size. Once this group size reaches a certain threshold (let’s call this “compaction threshold”), then the process merges these tables and flushes this new SS table in the lower level.

Problem 1 : 50% of disc should be available for compaction

During compaction, enough space is required for the original SS tables, and the destination compacted SS table. Therefore in the worst case scenario, 50% of disc is filled up with original SS tables, and those when being compacted need 50% more space for the destination compacted SS table.

Problem 2 : Data is scattered across levels and SS tables

Updates lead to stale data being present inside various SS tables inside different levels in disc. Lower levels will have a lot of stale data, because recent updates reach to lower levels after a lot of compactions.

How is write amplification equal to $O(log_{m}​N)$?

Let’s say initially we have N SSTables. Now, let’s say the min tables for merging is 2. In this case, you’ll need $O(log_{2}​N)$ operations to reach a single SSTable.

Operations:

• $N$ → $N/2$ SS tables : 1 operation
• $N/2$ → $N/4$ SS tables : 2 operations
• $N/2^i$ → $N/2^{i+1}$ SS tables : $i$ operations
• $N/2^{j−1}$ → $N/2^j$ = 1 SS table : $j=log_{2}​N$ operations

Pros & Cons

Pros	Cons
Very good choice for write heavy workloads - low write amplification	Data is split across many SS tables even at the same level. Therefore reads are slow
	Lower levels have a lot of obsolete data, increasing space amplification significantly
	For x size of SS tables required to be compacted, 2x space is required for the compaction process

Strategy 2 : Leveled compaction

In leveled compaction, unlike size tiered compaction, we merge all the SS tables in a level in a sorted fashion, to create SS tables of the next level. Therefore, all the items inside a level get flushed into the next level, as part of this compaction.

There are two key attributes in question here, max SS table size, and max level size. Compaction itself is triggered, when max level size is reached. While compaction, merging of higher leveled SS tables is done to create lower level SS tables. Lower level SS tables of size max_ss_table_size are created. Once that size is reached, next ss table is created.

Each level size is a multiple of the previous level size. So if the level L0 has a size 160 MB, then L1 can have a size 1600 MB, L2 can have a size 16000 MB etc.

The special MemTable to L0 transfer

As data is written to mem table, the max size for the mem table is reached. This table is then flushed to the L0 level as is. Note that there is no compaction process when an item is moved from memory to the L0 level.

Once max level size of the L0 level is reached, then only the sorted run happens, which merges the SS tables inside L0 in a sorted fashion, to create SS tables inside L1.

How exactly is the compaction process happening?

Compaction is triggered when the max level size is reached. Now, we already have a list of SS tables in the higher level, which are required to be written into the lower level. Assuming that the SS tables themselves are sorted (current level is not L0), we will go one by one through the SS tables, and identify the key range. For this key range, we search valid SS tables inside the lower level. We merge this data, remove duplicates, and then create new SS tables. The older SS tables are removed.

Because of this de-duplication process, all keys inside a level are unique, and there are no duplicates.

Why does leveled compaction have a low read amplification?

Since the data inside each level is always kept in a sorted fashion, reading inside each level is very optimized. Once a key is found inside a level, you don’t need to search a lower level, or worry about duplicates of the same key inside the same level.

Why does leveled compaction have a low space amplification?

The temporary space problem

The SS tables in the case of leveled compaction remain of the same size (160 MB generally). Therefore, during compaction we don’t need a lot of temporary space unlike size tiered compaction.

To be exact we will have 10 SS tables of a lower level, which we are compacting in a sorted fashion, and 1 SS table to which we are writing. Once the 1 output SS table is written, we can write it to the next level, and then start working on creating the next SS table from the 10 SS tables of the lower level.

Therefore, we will only need 11 * SS table of temporary disc space, 160 * 11 = 1760 MB < 2 GB

The duplicate data problem

Since there is an exponential grown of level sizes, as we move from higher levels to lower levels, the majority of data in the case of leveled compaction, is in the lower most levels. In the case we have three levels, L0, L1 and L2, and L0 has x size, L1 has m×x size and L2 will have m2×x size. m mostly is 10, so $x,10\times x,100\times x$, therefore the amount of data inside the last level (L2) = $\frac{100 \times x}{111 \times x}$​ which is equal to 90%

Now when data is pushed from L1 to L2, it’s sorted. Because of this, there can not be any duplicates inside a single level. Therefore, 90% of the data can not have any duplicates whatsoever.

Why does leveled compaction have a high write amplification?

During compaction, a single item might be written multiple times, during movement from a higher level to a lower level, along with in the case of overlapping key ranges, the lower level SS table might get re-written into new SS tables. Therefore, leveled compaction has a higher write amplification as compared to size tiered compaction.

Pros & Cons

Pros	Cons
Low space amplification	High write amplification
Low read amplification

Used by?

ScyllaDB and Apache Cassandra

Strategy 3 : [Bonus] Time window compaction

<TODO>

Appendix

Resources

• Bloom filter : https://brilliant.org/wiki/bloom-filter/
• DS210.15 Leveled Compaction | Operations with Apache Cassandra : https://www.youtube.com/watch?v=6yJEwqseMY4
• DS210.16 Size-Tiered Compaction | Operations with Apache Cassandra : https://www.youtube.com/watch?v=TyTXOjFMi7k
• DS210.17 Time Window Compaction | Operations with Apache Cassandra : https://youtu.be/P8jPlppa_8I?si=dtSCJ5WW_d9II36W
• Compaction : https://github.com/facebook/rocksdb/wiki/Compaction
• Choose a Compaction Strategy : https://opensource.docs.scylladb.com/stable/architecture/compaction/compaction-strategies.html#id1

 1295D - Same GCDs

Link to the problem : https://codeforces.com/problemset/problem/1295/D

I had to see the tutorial to understand this.

$$ gcd(a, m) = gcd(a + x, m)$$

Now, using Euclid's algorithm, we can say that 

$$ gcd(a, b) = gcd(a - b, b)$$

So, in our case, that would be:

$$ gcd(a + x, m) = gcd(a + x - m, m) = gcd(a + x - 2 \times m, m) = ...  = gcd((a + x) \% m, m)$$

Now, in the problem it's given that $ 0 \leq x < m$, so we can say that $ a \leq a + x < a + m$, and therefore we can also say that $ 0 \leq (a + x)\%m < m$

So, we can assume $$ Y = (a + x)\%m $$

We can say:

$$ gcd(a, m) = gcd(a + x, m) = gcd(Y, m)$$

Let's say that $gcd(a, m) = G$

So $$ gcd(Y, m) =  G$$

This means that G can divide both Y, and m, and there are no further common divisors (besides 1) in Y and m. This means:

$$ gcd(Y / G, m / G) = 1$$

Now notice that the variable here is $Y/G$. We want it to be such that it's co-prime to $m/G$. Also notice that since $Y < m$, $Y/G < m/G$. Therefore the Euler's totient function is applicable here. More details on that : https://cp-algorithms.com/algebra/phi-function.html

So we just need to calculate the totient function for $m/G = m / gcd(a, m)$

My submission : https://codeforces.com/contest/1295/submission/240298664