The proposal of Timestamp Ordering Transactions Over RocksDB

# The proposal of Timestamp Ordering Transactions Over RocksDB

by @riversand963 & @wolfkdy

## Abstraction

In this article, we introduce the timestamp ordering transactions over RocksDB. It's a cooperation of three layers, the application-layer commits a transaction with an user-defined timestamp, certainly, the timestamp has to satisfy several restrictions. The transaction layer, which mainly handles write-write conflict check. The rocksdb layers, we augmented the internalkey with an extra commit-timestamp, Necessary modifications have to be added into read/write/flush/iterate/compaction paths to support the correct semantics.

## Introduction

Timestamp Ordering(T/O for short) transaction is a rising star in industrial database circles these two years. Although it has been academicly investigated since the 1980s, its acceptance and popular use starts not long ago. CockroachDB[2], based on rocksdb, keeps a timestamp-based version-meta for every single key. Thus each read or write has to take extra cost to maintain the meta data. MongoDB4.0, based on wiredTiger3.0, fulfills T/O txns in an more elegant way[3], wiredTiger encodes timestamp in each key, The internal processes, such as version garbage collection, visibility check, have to consider both timestamps and snapshots.

Base on @riversand963 's current work about timestamps on RocksDB, I think it's time to make a conclusion, and move on to another greater stage. We proposal the t/o txn manager over RocksDB in this article, In chapter 1, We will describe this functionality with a few code pieces. In this way, a brief impression can be gained. In the later chapters, we will show what is added into the rocksdb codebase to support this functionality. then, we will describe several restrictions that user-level has to obey.

## 1. Basic APIs

Timestamp Ordering Transaction provides these apis below, they are quite alike the tranditional transaction, except two timestamp-related apis. SetReadTimeStamp provides the ability of time-travel read. SetCommitTimeStamp provides the ability of user-defined asif-commit-timestamp.

```c++
class TOTransaction {
public:
Status SetCommitTimeStamp(const RocksTimeStamp& timestamp);
Status GetReadTimeStamp(RocksTimeStamp& timestamp) const;
Status Commit();
Status Rollback();
Status Get(ReadOptions& options, const Slice& key, std::string& value);
Iterator GetIterator(ReadOptions& read_options);
Status Put(const Slice& key, const Slice& value);
Status Delete(const Slice& key);
}
```

ReadTimeStamp and CommitTimestamp are the magical part of our work. They provide an extra visibility check beyond rocksdb's basic SI check. First and most important, the SI rule will never be violated in our T/O txns, besides that, the timestamp order between different transactions also matters. We will show the rules with a few examples.

```c++
1 TEST_F(TOTransactionTest, ValidateIsolation) {
2 WriteOptions write_options;
3 ReadOptions read_options;
4 string value;
5 Status s;
6 TOTransaction* txn = txn_db->BeginTransaction(write_options, txn_options);
7 TOTransaction* txn2 = txn_db->BeginTransaction(write_options, txn_options);
8 ASSERT_TRUE(txn->GetID() < txn2->GetID());
9 ASSERT_OK(txn->SetReadTimeStamp(50, 0));
10 ASSERT_OK(txn->Put(Slice("A"), Slice("A-A")));
11 ASSERT_OK(txn->SetCommitTimeStamp(100));
12 ASSERT_OK(txn->Commit());
13 ASSERT_OK(txn2->Put(Slice("B"), Slice("B-B")));
14 ASSERT_OK(txn2->SetCommitTimeStamp(110));
15 TOTransaction* txn3 = txn_db->BeginTransaction(write_options, txn_options);
16 TOTransaction* txn4 = txn_db->BeginTransaction(write_options, txn_options);
17 ASSERT_OK(txn3->SetReadTimeStamp(60, 0));
18 ASSERT_OK(txn4->SetReadTimeStamp(110, 0));
19 s = txn3->Get(read_options, "A", &value);
20 ASSERT_TRUE(s.IsNotFound());
21 s = txn4->Get(read_options, "A", &value);
22 ASSERT_OK(s);
23 ASSERT_EQ(value, "A-A");
24 s = txn3->Get(read_options, "B", &value);
25 ASSERT_TRUE(s.IsNotFound());
26 s = txn4->Get(read_options, "B", &value);
27 ASSERT_TRUE(s.IsNotFound());
28 s = txn2->Commit();
29 ASSERT_OK(s);
30 s = txn3->Get(read_options, "B", &value);
31 ASSERT_TRUE(s.IsNotFound());
32 s = txn4->Get(read_options, "B", &value);
33 ASSERT_TRUE(s.IsNotFound());
```

In line 12, txn commits with commitTimestamp = 100, in line 15 and 16, txn3 and txn4 starts, with the rule of SI, txn3 and txn4 both can see the commit by txn, but the different readTimestamps of txn3 and txn4 set in line 17 and line 18 makes the result different. Txn3 can not see the modification by txn, but txn4 can see it. This is because txn3.readTimestamp < txn.commitTimestamp < txn4.readTimestamp.
In line 24 and 26, txn3 and txn4 can not see the modification by txn2 on B because B has not yet committed, they can not see it due to the rule of SI(does not read uncommitted data). However, in line 28, txn2 commits, in line 30 and 32, txn3 and txn4 still can not see it, this is still due to SI(does not read data committed after my start).

The formal visibility-check and write-write conflict check rule will be discussed in later chapters. In the next chapters, we will discuss each layer of our modification, from the bottommost layer to the application layer.

## 2. An overview of rocksdb layer's modification

There are five core parts that we have to make correct. They are the read path, write path, iteration path, compaction path and flushing path. The foundation underneath them all is the new format of the internal key.

### Internalkey Format

We put a 8 byte int after the userKey, before the lsn. The format is presented below:

```
UserKey + CommitTimestamp(8byte) + LSN + Type
```

As we will discuss in later chapters, it is guaranteed that to the same key, the lsn order is the same as its commitTimestamp order. The comparation of different versions to the same key can done on either LSN or commitTimestamp, and both will get the same result.

### Write Path

Every transaction preserves a writebatch, on commit, the writebatch will be written into memtable by DBImpl::WriteImpl. In this function, we iterate the writebatch and rewrite every userkey, append a timestamp after it. Actually, we insert the rewrited writebatch into memtable.

### Read Path
#### Memtable::SaveValue

If two internal keys have the same userkey, we should contine the search process iff

+ The lsn of the key is larger than the readOption's snapshot lsn.
+ The commitTimestamp of the key is larger than the readOption's readTimestamp

#### BlockBasedTable::Get

It should have similar logic as Memtable::SaveValue

### IterationPath

DBIter::IsVisible takes readOption's snapshot sequence and readTimestamp into consideration, has the similar logic as Memtable::SaveValue

### CompactionPath

Application level decides an **oldestTimestamp**, the oldestTimestamp is a database scope variable. Application level readTimestamp and commitTimestamp must not be older than oldestTimestamp. key versions with timestamps greater than oldestTimestamp are not allowed to be purged by flush or compaction procedure because these versions may be required by a timestamped-read in the future.

### CompactionIterator::NextFromInput

We should check at the beginning of CompactionIterator::NextFromInput if the incoming key has a timestamp larger than the oldestTimestamp. If so, we should keep the key version and continue to the next key.

## 3. The Timestamp Ordering Transaction Layer

The new feature supported by the rocksdb layer is only a part of the whole blueprint. Another transaction layer for timestamp based conflict-check, visibility-check, and support of some timestamp meta query and setting is also necessary.
```c++
class TOTransactionDB : public StackableDB {
public:
static Status Open(const Options& options,
const TOTransactionDBOptions& txn_db_options,
const std::string& dbname, TOTransactionDB** dbptr);

TOTransaction* BeginTransaction(
const WriteOptions& write_options,
const TOTransactionOptions& txn_options) = 0;

Status SetTimeStamp(const TimeStampType& ts_type, const RocksTimeStamp& ts);

Status QueryTimeStamp(const TimeStampType& ts_type, RocksTimeStamp* timestamp);
};

```
Despite the basic features that every transactional database have, we provide two extra api, by**SetTimestamp**, the application-level can communicate with the transaction layer which timestamp is the oldest to maintain, By **QueryTimestamp**, the **pinned-timestamp**, that says, the minimum of all **readTimestamp** and **oldestTimestamp** can be returned.

+ pinned-timestamp = min(min(readTimestamps), oldestTimestamp)

Another timestamp, **allcommitted-timestamp** can also be queried by this api. **allcommitted-timestamp** is a more complicated concept, which is better to look into the code for understanding.

TOTransactionDB basicly follows the first-update-wins occ strategy for conflict-check, which in some circumstances may lead to false-postive write-write conflicts, but gains at least three times throughput than rocksdb's optimistic transaction. The conflict check prodecure is described below:

```
// uncommitted keys is a db-scope global string set.
uncommittedKeys: set

Transaction::Put(key) {
if (key in uncommittedKeys) return WWError
uncommittedKeys.add(key)
latestLsn, latest_committs = GetKeyFromDB(key)
if (latestLsn > this.snapshot.lsn) {
// disobeys SI
return WWError
}
if (latest_committs > this.readTs) {
// disobeys T/O
return WWError
}
this.batch.Put(key)
}
```
We maintain a set called uncommittedkeys to track all the keys that have not been committed, but is owned by one active transaction. the uncommittedkeys guarantees that a key can be processed by at most one txn at the same time. When a txn tries to modify some key, it first tries to own the key by adding the key to uncommittedkeys if possible. Then it retrieves the latest version of the key from the database. Here, the latest version means both the latestLsn and the latest_commit_timestamp, as we will show in the following chapters, to the same key, lsn order is the same as commit_timestamp order, it can be guaranteed that the first key we met during the seeking process of a userkey has both the latest lsn and the latest commit_timestamp. if latestLsn > this.snapshot.lsn, then the put by this transaction can not continue because it violates SI, if latest_committs > this.readTimestamp, then the put can not continue because it violates T/O.

## 4. Application layer's rules

Generally speaking, The commits to a particular key **MUST** be performed in timestamp order. There are many strategies to satisfy this restriction. The most easy and straightforwad one is to make transactions single-threaded, start and commit one by one. It is definitely not a good solution. An alternative is the PBMA strategy.

### 4.1 The Post-Begin-Monotonic-Allocation Strategy

We will introduce the PBMA strategy, which makes it possible for all tsTxns to run in parallel, but commit in commitTimestamp order.

```
1. // the global timestamp allocator
2. GTA -> {
3. lock Lock
4. globalTs int
5. }
6. // the timestamp allocation prodecure
7. Transaction txn
8. txn.Begin()
9. with(GTA.lock) {
10. txn.commitTimestamp = globalTs++
11.}
12.......
13.txn.Commit()
```
### 4.2 To the same key, PBMA guarantees commitTimestamp order the same as txnId order

SI guarantees that if two transactions overlap in their lifetime and try to modify the same key, at most one can success. So will there be some situations that a transaction with a greater commitTimestamp starts and commits first, while a transaction with a smaller commitTimestamp starts and commits later, so that they have no overlap? The answer is still no.Suppose two transactions txn1 and txn2 have no overlap, txn2 is earlier than txn1 and txn2 has a greater commitTimestamp(=2) than txn1(commitTimetamp=1). This can't happen, because in the code block above, line 9,10 guarantees that commitTimestamps are allocated within a lock, they should be monotonic relatively to the wallclock.