RocksDB is a popular lightweight embedded key-value database. Flink uses it in RocksDBStateBackend to store large state.

Background

RocksDB has proven itself in many deployments. As a general-purpose key-value database, however, it struggles with many Flink state-storage workloads. Flink currently offers no better alternative, but this article can help developers understand the issues they encounter with RocksDBStateBackend.

RocksDB

Let us first take a brief look at RocksDB.

RocksDB In Flink

RocksDB’s performance goals:

• Good performance for both point and range queries.
• Fast random reads.
• Fast updates.
• Tunability for different hardware.

Basic data structures:

• memtable: The in-memory storage structure. New writes enter it after being recorded in the WAL.
• sstfile: An on-disk storage file to which memtable data is flushed.
• logfile: Used to recover lost memtable data.

New data is first written to the logfile and then synchronously to the memtable. When a condition such as the memtable size limit is met, the data is persisted to disk. Periodic compaction across SST files reduces their number and combines operations on the same key. This should be familiar if you know HBase’s MemStore, HFile, and WAL: both use the LSM-tree approach.

Compaction

RocksDB provides three compaction styles:

• Level Style Compaction (default): SST files are organized into levels, with older files at higher levels. Smaller files are periodically merged into the next level. This improves queries and reduces read amplification, with a multithreaded implementation.
• Universal Style Compaction: Merges similarly sized files to reduce write amplification.
• FIFO Style Compaction: A relatively coarse implementation that can evict data.

Compaction is familiar in storage systems for real-time workloads. Cassandra and HBase use similar ideas.

Index Sstfile

With default level-style compaction, RocksDB contains multiple levels of sorted arrays. Locating an SST file is therefore an interesting problem: for a point lookup at Level N, how can it quickly find the appropriate file?

Fractional Cascading

It uses fractional cascading. This is still fundamentally binary search, but the search interval shrinks as the lookup moves through successive levels.

For example:

1
L1 = [0 - 0], [1 - 5], [6 - 10], [11 - 15]
2
L2 = [0 - 1], [2 - 3], [4 - 12], [13 - 18]
3
L3 = [0 - 7], [8 - 14], [15 - 19], [20 - 30]

As shown, each SST file covers a range. Independently binary-searching every level would be inefficient. Fractional cascading stores indexes into the next level for each SST file, progressively narrowing the search. With these indexes, the layout becomes:

1
L1 = [[0 - 0], 0], [[1 - 5], 2], [[6 - 10], 2], [[11 - 15], 3]
2
L2 = [[0 - 1], 0], [[2 - 3], 0], [[4 - 12], 1], [[13 - 18], 3]
3
L3 = [[0 - 7], 0], [[8 - 14], 0], [[15 - 19], 0], [[20 - 30], 0]

Suppose we look up key ‘8’. At L1, the relevant SST file covers [6 - 10]. Its L2 index is 2, so the L2 search can be restricted to indexes [0, 2] instead of searching the entire level.

Transaction

RocksDB divides transactions into three stages: Put, Prepare, and Commit. Waiting until every operation has been prepared and only persisting to disk at Commit would hurt throughput when several large transactions run concurrently.

There are two improvements:

• WritePrepared: Persist to disk during Prepare.
• WriteUnprepared: Persist to disk during Put.

WritePrepared

RocksDB adds a prepare_seq to each record. After a transaction commits, CommitCache stores a mapping between prepare_seq and commit_seq. CommitCache evicts entries when it reaches its size limit; max_evict_seq represents the most recently evicted prepare_seq.

During a query, once a record is found, its prepare_seq determines visibility:

• prepare_seq > max_evict_seq and absent from CommitCache: The record is still prepared but uncommitted, so it cannot be read.
• prepare_seq <= max_evict_seq and prepare_seq absent from CommitCache: The record has been persisted and can be read directly.
• prepare_seq in CommitCache or prepare_seq < max_evict_seq: The record has committed and can be read directly.

There are special cases. If a transaction takes too long between Prepare and Commit and max_evict_seq passes the record’s prepare_seq, then prepare_seq is placed in the delayed_prepared_ set. RocksDB checks this set on each query to determine whether the record has committed.

WriteUnprepared

I have not fully understood this mechanism from the documentation alone, and the feature has not yet been used in production…

After reading the RocksDB documentation, I suspect many developers would hesitate to depend on its transactions. First, the algorithm is complicated and difficult to understand, which would make production debugging challenging. Second, it does not handle certain extremes well, such as long-running transactions or reading very old snapshots. Its optimizations to Put -> Prepare -> Commit improve throughput, but impose substantial constraints on applications.

Summary

This is the second note in my Database series. I chose RocksDB because many companies have used it successfully, and we considered it at work too (although throughput did not meet our expectations). Most information here comes from the RocksDB Wiki. The architecture offers few surprises, and transaction handling leaves something to be desired. In use, though, RocksDB is genuinely lightweight: no separate deployment is required, a single JAR provides a complete database, the API is pleasant, and HDFS integration is flexible.

Academic work has proposed ways to address RocksDB’s read and write amplification. I have briefly studied some of it and found the perspective interesting, but I wonder whether there are successful implementations.

• WiscKey: Separating Keys from Values in SSD-conscious Storage
• HashKV: Enabling Efficient Updates in KV Storage via Hashing