At the highest level, toyDB consists of a cluster of nodes that execute SQL transactions against a replicated state machine. Clients can connect to any node in the cluster and submit SQL statements. It aims to provide linearizability (i.e. strong consistency) and serializability, but falls slightly short as it currently only implements snapshot isolation.
The Raft algorithm is used for cluster consensus, which tolerates the failure of any node as long as a majority of nodes are still available. One node is elected leader, and replicates commands to the others which apply them to local copies of the state machine. If the leader is lost, a new leader is elected and the cluster continues operation. Client commands are automatically forwarded to the leader.
This architecture guide will begin with a high-level overview of node components, before discussing each component from the bottom up. Along the way, we will make note of tradeoffs and design choices.
A toyDB node consists of three main components:
-
Storage engine: stores data and manages transactions, on disk and in memory.
-
Raft consensus engine: handles cluster coordination and state machine replication.
-
SQL engine: parses, plans, and executes SQL statements for clients.
These components are integrated in the toyDB server, which handles network communication with clients and other nodes. The following diagram illustrates its internal structure:
At the bottom is a simple key/value store, which stores all SQL data. This is wrapped inside an MVCC key/value store that adds ACID transactions. On top of that is a SQL storage engine, providing basic CRUD operations on tables, rows, and indexes. This makes up the node's core storage engine.
The SQL storage engine is wrapped in a Raft state machine interface, allowing it to be managed by the Raft consensus engine. The Raft node receives commands from clients and coordinates with other Raft nodes to reach consensus on an ordered command log. Once commands are committed to the log, they are applied to the local state machine.
On top of the Raft engine is a Raft-based SQL storage engine, which implements the SQL storage interface and submits commands to the Raft cluster. This allows the rest of the SQL layer to use the Raft cluster as if it was local storage. The SQL engine manages client SQL sessions, which take SQL queries as text, parse them, generate query plans, and execute them against the SQL storage engine.
Surrounding these components is the toyDB server, which in addition to network communication also handles configuration, logging, and other process-level concerns.
ToyDB uses a pluggable key/value storage engine, with the SQL and Raft storage engines configurable
via the storage_sql and storage_raft options respectively. The higher-level SQL storage engine
will be discussed separately in the SQL section.
A key/value storage engine stores arbitrary key/value pairs as binary byte slices, and implements
the
storage::Engine
trait:
/// A key/value storage engine, where both keys and values are arbitrary byte
/// strings between 0 B and 2 GB, stored in lexicographical key order. Writes
/// are only guaranteed durable after calling flush().
///
/// Only supports single-threaded use since all methods (including reads) take a
/// mutable reference -- serialized access can't be avoided anyway, since both
/// Raft execution and file access is serial.
pub trait Engine: std::fmt::Display + Send + Sync {
/// The iterator returned by scan(). Traits can't return "impl Trait", and
/// we don't want to use trait objects, so the type must be specified.
type ScanIterator<'a>: DoubleEndedIterator<Item = Result<(Vec<u8>, Vec<u8>)>> + 'a
where
Self: 'a;
/// Deletes a key, or does nothing if it does not exist.
fn delete(&mut self, key: &[u8]) -> Result<()>;
/// Flushes any buffered data to the underlying storage medium.
fn flush(&mut self) -> Result<()>;
/// Gets a value for a key, if it exists.
fn get(&mut self, key: &[u8]) -> Result<Option<Vec<u8>>>;
/// Iterates over an ordered range of key/value pairs.
fn scan<R: std::ops::RangeBounds<Vec<u8>>>(&mut self, range: R) -> Self::ScanIterator<'_>;
/// Sets a value for a key, replacing the existing value if any.
fn set(&mut self, key: &[u8], value: Vec<u8>) -> Result<()>;
}The get, set and delete methods simply read and write key/value pairs, and flush ensures
any buffered data is written out to storage (e.g. via the fsync system call). scan iterates
over a key/value range in order, a property that is crucial to higher-level functionality (e.g.
SQL table scans) and has a couple of important implications:
-
Implementations should store data ordered by key, for performance.
-
Keys should use an order-preserving byte encoding, to allow range scans.
The engine itself does not care what keys contain, but the storage module offers an order-preserving key encoding called KeyCode for use by higher layers. These storage layers often use composite keys made up of several possibly variable-length values (e.g. an index key consists of table, column, and value), and the natural ordering of each segment must be preserved, a property satisfied by this encoding:
bool:0x00forfalse,0x01fortrue.u64: big-endian binary representation.i64: big-endian binary representation, with sign bit flipped.f64: big-endian binary representation, with sign bit flipped, and rest if negative.Vec<u8>:0x00is escaped as0x00ff, terminated with0x0000.String: likeVec<u8>.
Additionally, several container types are supported:
- Tuple: concatenation of elements, with no surrounding structure.
- Array: like tuple.
- Vec: like tuple.
- Enum: the variant's enum index as a single
u8byte, then contents. - Value: like enum.
The default key/value engine is
storage::BitCask,
a very simple variant of Bitcask, an append-only log-structured storage engine.
All writes are appended to a log file, with an index mapping live keys to file
positions maintained in memory. When the amount of garbage (replaced or deleted
keys) in the file exceeds 20%, a new log file is written containing only live
keys, replacing the old log file.
Keyset in memory: BitCask requires the entire key set to fit in memory, and must also scan the log file on startup to construct the key index.
Compaction volume: unlike an LSM tree, this single-file BitCask implementation requires rewriting the entire dataset during compactions, which can produce significant write amplification over time.
Key encoding: does not make use of any compression, e.g. variable-length integers, preferring simplicity and correctness.
MVCC (Multi-Version Concurrency Control) is a relatively simple concurrency control mechanism that provides ACID transactions with snapshot isolation without taking out locks or having writes block reads. It also versions all data, allowing querying of historical data.
toyDB implements MVCC at the storage layer as
storage::mvcc::MVCC,
using any storage::Engine implementation for underlying storage. begin returns a new
transaction, which provides the usual key/value operations such as get, set, and scan.
Additionally, it has a commit method which persists the changes and makes them visible to
other transactions, and a rollback method which discards them.
When a transaction begins, it fetches the next available version from Key::NextVersion and
increments it, then records itself as an active transaction via Key::TxnActive(version). It also
takes a snapshot of the active set, containing the versions of all other active transactions as of
the transaction start, and saves it as Key::TxnActiveSnapshot(id).
Key/value pairs are saved as Key::Version(key, version), where key is the user-provided key
and version is the transaction's version. The visibility of key/value pairs for a transaction is
given as follows:
-
For a given user key, do a reverse scan of
Key::Version(key, version)starting at the current transaction's version. -
Skip any records whose version is in the transaction's snapshot of the active set.
-
Return the first matching record, if any. This record may be either a
Some(value)or aNoneif the key was deleted.
When writing a key/value pair, the transaction first checks for any conflicts by scanning for a
Key::Version(key, version) which is not visible to it. If one is found, a serialization error
is returned and the client must retry the transaction. Otherwise, the transaction writes the new
record and keeps track of the change as Key::TxnWrite(version, key) in case it must roll back.
When the transaction commits, it simply deletes its Txn::Active(id) record, thus making its
changes visible to any subsequent transactions. If the transaction instead rolls back, it
iterates over all Key::TxnWrite(id, key) entries and removes the written key/value records before
removing its Txn::Active(id) entry.
This simple scheme is sufficient to provide ACID transaction guarantees with snapshot isolation: commits are atomic, a transaction sees a consistent snapshot of the key/value store as of the start of the transaction, and any write conflicts result in serialization errors which must be retried.
To satisfy time travel queries, a read-only transaction simply loads the Key::TxnActiveSnapshot
entry of a past transaction and applies the same visibility rules as for normal transactions.
Serializability: snapshot isolation is not fully serializable, since it exhibits write skew anomalies. This would require serializable snapshot isolation, which was considered unnecessary for a first version - it may be implemented later.
Garbage collection: old MVCC versions are never removed, leading to unbounded disk usage. However, this also allows for complete data history, and simplifies the implementation.
Transaction ID overflow: transaction IDs will overflow after 64 bits, but this is never going to happen with toyDB.
The Raft consensus protocol is explained well in the original Raft paper, and will not be repeated here - refer to it for details. toyDB's implementation follows the paper fairly closely.
The Raft node raft::Node is
the core of the implementation, a finite state machine with enum variants for the node roles:
leader, follower, and candidate. This enum wraps the RawNode struct, which contains common
node functionality and is generic over the specific roles Leader, Follower, and Candidate
that implement the Raft protocol.
Nodes are initialized with an ID and a list of peer IDs, and communicate by passing
raft::Message
messages. Inbound messages are received via Node.step() calls, and outbound messages are sent
via an mpsc channel. Nodes also use a logical clock to keep track of e.g. election timeouts
and heartbeats, and the clock is ticked at regular intervals via Node.tick() calls. These
methods are synchronous and may cause state transitions, e.g. changing a candidate into a leader
when it receives the winning vote.
Nodes have a command log raft::Log,
using a storage::Engine for storage, and a raft::State
state machine (the SQL engine). When the leader receives a write request, it appends the command
to its local log and replicates it to followers. Once a quorum have replicated it, the command is
committed and applied to the state machine, and the result returned the the client. When the leader
receives a read request, it needs to ensure it is still the leader in order to satisfy
linearizability (a new leader could exist elsewhere resulting in a stale read). It increments a
read sequence number and broadcasts it via a Raft heartbeat. Once a quorum have confirmed the
leader at this sequence number, the read command is executed against the state machine and the
result returned to the client.
The actual network communication is handled by the server process, which will be described in a separate section.
Single-threaded state: all state operations run in a single thread on the leader, preventing horizontal scalability. Improvements here would require running multiple sharded Raft clusters, which is out of scope for the project.
Synchronous application: state machine application happens synchronously in the main Raft thread. This is significantly simpler than asynchronous application, but may cause delays in Raft processing.
Log replication: only the simplest form of Raft log replication is implemented, without state snapshots or rapid log replay. Lagging nodes will be very slow to catch up.
Cluster resizing: the Raft cluster consists of a static set of nodes given at startup, resizing it requires a complete cluster restart.
The SQL engine builds on Raft and MVCC to provide a SQL interface to clients. Logically, the life of a SQL query is as follows:
Query → Lexer → Parser → Planner → Optimizer → Executor → Storage Engine
We'll begin by looking at the basic SQL type and schema systems, as well as the SQL storage engine and its session interface. Then we'll switch sides and look at how a query is executed, starting at the front with the parser and following it until it's executed against the SQL storage engine, completing the chain.
toyDB has a very simple type system, with the
sql::DataType enum
specifying the available data types: Boolean, Integer, Float, and String.
The sql::Value enum
represents a specific value using Rust's native type system, e.g. an integer value is
Value::Integer(i64). This enum also specifies comparison, ordering, and formatting of values. The
special value Value::Null represents an unknown value of unknown type, following the rules of
three-valued logic.
Values can be grouped into a Row, which is an alias for Vec<Value>. The type Rows is an alias
for a fallible row iterator, and Column is a result column containing a name.
Expressions sql::Expression
represent operations on values. For example, (1 + 2) * 3 is represented as:
Expression::Multiply(
Expression::Add(
Expression::Constant(Value::Integer(1)),
Expression::Constant(Value::Integer(2)),
),
Expression::Constant(Value::Integer(3)),
)Calling evaluate() on the expression will recursively evaluate it, returning Value::Integer(9).
The schema defines the tables sql::Table
and columns sql::Column
in a toyDB database. Tables have a name and a list of columns, while a column has several
attributes such as name, data type, and various constraints. They also have methods to
validate rows and values, e.g. to make sure a value is of the correct type for a column
or to enforce referential integrity.
The schema is stored and managed with sql::Catalog,
a trait implemented by the SQL storage engine:
pub trait Catalog {
/// Creates a new table.
fn create_table(&mut self, table: &Table) -> Result<()>;
/// Deletes a table, or errors if it does not exist.
fn delete_table(&mut self, table: &str) -> Result<()>;
/// Reads a table, if it exists.
fn read_table(&self, table: &str) -> Result<Option<Table>>;
/// Iterates over all tables.
fn scan_tables(&self) -> Result<Tables>;
}Single database: only a single, unnamed database is supported per toyDB cluster. This is sufficient for toyDB's use-cases, and simplifies the implementation.
Schema changes: schema changes other than creating or dropping tables is not supported. This avoids complicated data migration logic, and allows using table/column names as storage identifiers (since they can never change) without any additional indirection.
The SQL storage engine trait is sql::Engine:
pub trait Engine: Clone {
type Transaction: Transaction;
/// Begins a transaction in the given mode.
fn begin(&self, mode: Mode) -> Result<Self::Transaction>;
/// Resumes an active transaction with the given ID.
fn resume(&self, id: u64) -> Result<Self::Transaction>;
/// Begins a SQL session for executing SQL statements.
fn session(&self) -> Result<Session<Self>> {
Ok(Session { engine: self.clone(), txn: None })
}
}The main use of the trait is to dispense sql::Session instances, individual client sessions which
execute SQL queries submitted as plain text and track transaction state. The actual storage
engine functionality is exposed via the sql::Transaction trait, representing an ACID transaction
providing basic CRUD (create, read, update, delete) operations for tables, rows, and indexes:
pub trait Transaction: Catalog {
/// Commits the transaction.
fn commit(self) -> Result<()>;
/// Rolls back the transaction.
fn rollback(self) -> Result<()>;
/// Creates a new table row.
fn create(&mut self, table: &str, row: Row) -> Result<()>;
/// Deletes a table row.
fn delete(&mut self, table: &str, id: &Value) -> Result<()>;
/// Reads a table row, if it exists.
fn read(&self, table: &str, id: &Value) -> Result<Option<Row>>;
/// Scans a table's rows, optionally filtering by the given predicate expression.
fn scan(&self, table: &str, filter: Option<Expression>) -> Result<Scan>;
/// Updates a table row.
fn update(&mut self, table: &str, id: &Value, row: Row) -> Result<()>;
/// Reads an index entry, if it exists.
fn read_index(&self, table: &str, column: &str, value: &Value) -> Result<HashSet<Value>>;
/// Scans a column's index entries.
fn scan_index(&self, table: &str, column: &str) -> Result<IndexScan>;
}The main SQL storage engine implementation is
sql::engine::KV, which
is built on top of an MVCC key/value store and its transaction functionality.
The Raft SQL storage engine
sql::engine::Raft
uses a Raft API client raft::Client to submit state machine commands specified by the enums
Mutation and Query to the local Raft node. It also provides a Raft state machine
sql::engine::raft::State which wraps a regular sql::engine::KV SQL storage engine and applies
state machine commands to it. Since the Raft SQL engine implements the sql::Engine trait, it can
be used interchangably with the local storage engine.
Raft result streaming: result streaming is not implemented for Raft commands, so the Raft SQL engine must buffer the entire result set in memory and serialize it before returning it to the client - particularly expensive for table scans. Implementing streaming in Raft was considered out of scope for the project.
The SQL session sql::Session
takes plain-text SQL queries via execute() and returns the result. The first step in this process
is to parse the query into an abstract syntax tree
(AST) which represents the query semantics. This happens as follows:
SQL → Lexer → Tokens → Parser → AST
The lexer
sql::Lexer takes
a SQL string, splits it into pieces, and classifies them as tokens sql::Token. It does not
care about the meaning of the tokens, but removes whitespace and tries to figure out if
something is a number, string, keyword, and so on. It also does some basic pre-processing, such as
interpreting string quotes, checking number formatting, and rejecting unknown keywords.
For example, the following input string results in the listed tokens, even though the query is invalid:
3.14 +UPDATE 'abc'→Token::Number("3.14")Token::PlusToken::Keyword(Keyword::Update)Token::String("abc")
The parser sql::Parser
iterates over the tokens generated by the lexer, interprets them, and builds an AST representing
the semantic query. For example, SELECT name, 2020 - birthyear AS age FROM people
results in the following AST:
ast::Statement::Select{
select: vec![
(ast::Expression::Field(None, "name"), None),
(ast::Expression::Operation(
ast::Operation::Subtract(
ast::Expression::Literal(ast::Literal::Integer(2020)),
ast::Expression::Field(None, "birthyear"),
)
), Some("age")),
],
from: vec![
ast::FromItem::Table{name: "people", alias: None},
],
where: None,
group_by: vec![],
having: None,
order: vec![],
offset: None,
limit: None,
}The parser will interpret the SQL syntax, determining the type of query and its parameters,
returning an error for any invalid syntax. However, it has no idea if the table people
actually exists, or if the field birthyear is an integer - that is the job of the planner.
Notably, the parser also parses expressions, such as 1 + 2 * 3. This is non-trivial due to
precedence rules, i.e. 2 * 3 should be evaluated first, but not if there are parentheses
around (1 + 2). The toyDB parser uses the
precedence climbing algorithm
for this. Also note that AST expressions are different from SQL engine expressions, and do not map
one-to-one. This is clearest in the case of function calls, where the parser does not know (or
care) if a given function exists, it just parses a function call as an arbitrary function name and
arguments. The planner will translate this into actual expressions that can be evaluated.
The SQL planner sql::Planner
takes the AST generated by the parser and builds a SQL execution plan
sql::Plan, which is an
abstract representation of the steps necessary to execute the query. For example, the following
shows a simple query and corresponding execution plan, formatted as EXPLAIN output:
SELECT id, title, rating * 100 FROM movies WHERE released > 2010 ORDER BY rating DESC;
Order: rating desc
└─ Projection: id, title, rating * 100
└─ Filter: released > 2010
└─ Scan: movies
The plan nodes sql::Node in a query execution plan represent a
relational algebra operator, where the
output from one node flows as input into the next. In the example above, the query first does a
full table scan of the movies table, then applies the filter released > 2010 to the rows,
before projecting (formatting) the result and sorting it by rating.
Most of the planning is fairly straightforward, translating AST nodes to plan nodes and expressions.
The trickiest part is resolving table and column names to result column indexes across multiple
layers of aliases, joins, and projections - this is handled with sql::plan::Scope, which keeps
track of what names are visible to the node being built and maps them to column indexes. Another
challenge is aggregate functions, which are implemented as a pre-projection of function
arguments and grouping/hidden columns, then an aggregation node, and finally a post-projection
evaluating the final aggregate expressions - like in the following example:
SELECT g.name AS genre, MAX(rating * 100) - MIN(rating * 100)
FROM movies m JOIN genres g ON m.genre_id = g.id
WHERE m.released > 2000
GROUP BY g.id, g.name
HAVING MIN(rating) > 7
ORDER BY g.id ASC;
Projection: #0, #1
└─ Order: g.id asc
└─ Filter: #2 > 7
└─ Projection: genre, #0 - #1, #2, g.id
└─ Aggregation: maximum, minimum, minimum
└─ Projection: rating * 100, rating * 100, rating, g.id, g.name
└─ Filter: m.released > 2000
└─ NestedLoopJoin: inner on m.genre_id = g.id
├─ Scan: movies as m
└─ Scan: genres as g
The planner generates a very naïve execution plan, primarily concerned with producing one that
is correct but not necessarily fast. This means that it will always do full table scans,
always use nested loop joins, and so on. The plan
is then optimized by a series of optimizers implementing
sql::Optimizer:
-
ConstantFolder: pre-evaluates constant expressions to avoid having to re-evaluate them for each row. -
FilterPushdown: pushes filters deeper into the query to reduce the number of rows evaluated by each node, e.g. by pushing single-table predicates all the way to the table scan node such that filtered nodes won't have to go across the Raft layer. -
IndexLookup: transforms table scans into primary key or index lookups where possible. -
NoopCleaner: attempts to remove noop operations, e.g. filter nodes that evaluate to a constantTRUEvalue. -
JoinType: transforms nested loop joins into hash joins for equijoins (equality join predicate).
Optimizers make heavy use of boolean algebra to transform expressions into forms that are more convenient to work with. For example, partial filter pushdown (e.g. across join nodes) can only push down conjunctive clauses (i.e. AND parts), so expressions are converted into conjunctive normal form first such that each part can be considered separately.
Below is an example of a complex optimized plan where table scans have been replaced with key and index lookups, filters have been pushed down into scan nodes, and nested loop joins have been replaced by hash joins. It fetches science fiction movies released since 2000 by studios that have released any movie with a rating of 8 or more:
SELECT m.id, m.title, g.name AS genre, m.released, s.name AS studio
FROM movies m JOIN genres g ON m.genre_id = g.id,
studios s JOIN movies good ON good.studio_id = s.id AND good.rating >= 8
WHERE m.studio_id = s.id AND m.released >= 2000 AND g.id = 1
ORDER BY m.title ASC;
Order: m.title asc
└─ Projection: m.id, m.title, g.name, m.released, s.name
└─ HashJoin: inner on m.studio_id = s.id
├─ HashJoin: inner on m.genre_id = g.id
│ ├─ Filter: m.released > 2000 OR m.released = 2000
│ │ └─ IndexLookup: movies as m column genre_id (1)
│ └─ KeyLookup: genres as g (1)
└─ HashJoin: inner on s.id = good.studio_id
├─ Scan: studios as s
└─ Scan: movies as good (good.rating > 8 OR good.rating = 8)
Type checking: expression type conflicts are only detected at evaluation time, not during planning.
Every SQL plan node has a corresponding executor, implementing the
sql::Executor trait:
pub trait Executor<T: Transaction> {
/// Executes the executor, consuming it and returning a result set
fn execute(self: Box<Self>, txn: &mut T) -> Result<ResultSet>;
}Executors are given a sql::Transaction to access the SQL storage engine, and return a
sql::ResultSet with the query result. Most often, the result is of type sql::ResultSet::Query
containing a list of columns and a row iterator. Most executors contain other executors that
they use as inputs, for example the Filter executor will often have a Scan executor as a source:
pub struct Filter<T: Transaction> {
source: Box<dyn Executor<T>>,
predicate: Expression,
}Calling execute on a sql::Plan will build and execute the root node's executor, which in
turn will recursively call execute on its source executors (if any) and process their results.
Executors typically augment the source's returned row iterator using Rust's
Iterator functionality, e.g. by
calling filter() on it and returning a new iterator. The entire execution engine thus works in
a streaming fashion and leverages Rust's zero-cost iterator
abstractions.
Finally, the root ResultSet is returned to the client.
The toyDB Server manages
network traffic for the Raft and SQL engines, using the Tokio async executor.
It opens TCP listeners on port 9605 for SQL clients and 9705 for Raft peers, both using
length-prefixed Bincode-encoded message passing via
Serde-encoded Tokio streams as a protocol.
The Raft server is split out to raft::Server,
which runs a main event loop routing Raft messages
between the local Raft node, TCP peers, and local state machine clients (i.e. the Raft SQL engine
wrapper), as well as ticking the Raft logical clock at regular intervals. It spawns separate Tokio
tasks that maintain outbound TCP connections to all Raft peers, while internal communication
happens via mpscchannels.
The SQL server spawns a new Tokio task for each SQL client that connects, running a separate
SQL session from the SQL storage engine on top of Raft. It communicates with the client by passing
server::Request and server::Response messages that are translated to sql::Session calls.
The main toydb binary
simply initializes a toyDB server based on command-line arguments and configuration files, and then
runs it via the Tokio runtime.
Security: all network traffic is unauthenticated an in plaintext, as security was considered out of scope for the project.
The toyDB Client provides a
simple API for interacting with a server, mainly by executing SQL statements via execute()
returning sql::ResultSet.
The toysql command-line
client is a simple REPL client that connects to a server using the toyDB Client and continually
prompts the user for a SQL query to execute, displaying the returned result.