Paper Notes: [OSDI'10] Large-scale Incremental Processing Using Distributed Transactions and Notifications

First, let’s review Bigtable’s data model:

As shown above, data can be identified by the following hierarchy:

Bigtable supports row-level transactions. Although [2] does not explicitly state the transaction isolation level, from the discussion in [1] about locking the primary, we can infer that it should be Serializability.

[3] informally describes Snapshot Isolation. In brief, a transaction under Snapshot Isolation has two important time points: Start-Timestamp and Commit-Timestamp. When the transaction starts, that is, at the Start-Timestamp, it takes a snapshot of the data in the entire system. All data read during the transaction and all changes written by the transaction come from this snapshot. At commit time, that is, at the Commit-Timestamp, if another transaction has already committed, this transaction aborts; otherwise, it succeeds.

Formal descriptions of Snapshot Isolation can be found in [4], [5], and [6].

There are many related topics around Snapshot Isolation, including the following major areas:

For related topics, see: Snapshot Isolation Survey, Paper Notes: [CONCUR'15] A Framework for Transactional Consistency Models with Atomic Visibility, and CAP, ACID, What Can We Do?.

P.S. For the implementation of MVCC in Postgres, see [10]. [3] mentions that Snapshot Isolation is an extension of the Multiversion Mixed Method in [7]. However, Snapshot Isolation can also be implemented in other ways. For example, [11] provides an approach that implements Snapshot Isolation using read-write locks.

Assume there is a Timestamp Oracle Service that can provide monotonically increasing timestamps. Bigtable Transaction provides strongly consistent transactions within a row. Bigtable uses timestamps provided by the Timestamp Oracle Service as cell version numbers. Based on these facilities, we introduce how Percolator Transaction is implemented.

In the process above, both read requests and write requests require support from Bigtable row-level transactions.

Because BigTable does not support true row-level locks, the lock here is implemented using a special column: c:lock (the : here can easily be ambiguous. In BigTable, : separates ColumnFamily and ColumnName, so you can think of this as using another special symbol as the separator, such as cf:cn+lock). As a result, the original data also needs to change accordingly: by convention, the column c:data stores the real data and its different versions. Because we want to support cross-row transactions, when a Percolator transaction eventually commits, we also need a special column to mark the commit: c:write.

Percolator also makes some changes to avoid assigning a unique identifier to every Percolator transaction. When locking, it arbitrarily designates the cell of one write request as the primary; this lock is the primary lock. The other locks in the transaction are secondary locks and point to this primary lock. This design effectively uses the primary lock to uniquely identify the Percolator transaction, and uses BigTable’s reliable records to record and track the entire transaction. This record can be used during failure recovery to read the state of the entire transaction.

For the detailed Percolator algorithm, refer to the original paper. The following explains the example given in the paper.

As shown in Table 1, the initial state stores two rows of data: Bob and Joe. The last committed version for both is 5, and neither is currently locked. Looking at the data column gives their current state: Bob has \$10, and Joe has \$2. If there is a "valid" lock in the version range of the cell being read ("valid" is discussed later in failure recovery), the read must wait until the lock is released.

Now Bob wants to transfer \$7 to Joe. None of the read requests need to acquire locks; they query directly according to the snapshot at the current time, so they read the state in Table 1: Bob has \$10, and Joe has \$2. The transfer is then determined to be possible, and the write begins. All write requests are first buffered locally and are performed during the commit phase, which ensures that, during the commit phase, all cells that need to be locked are known. The transfer operation affects Bob’s bal and Joe’s bal respectively.

Assume we always treat the cell affected by the first write operation as the primary lock object. First, \$7 is subtracted from Bob’s account, so the first write can both acquire the primary lock and record the write request, as shown in Table 2.

There are two cases in which locking can fail:

After the primary lock succeeds, all subsequent write operations also need to acquire locks, and the locks point to the primary lock for failure detection and recovery.

If all write requests are locked successfully, the transaction can enter the commit phase. During commit, the content locked by the primary lock is committed first; see The transaction has now reached the commit point.

Then the content locked by the other locks is committed:

Percolator effectively uses the client to coordinate, but persists all required data into Bigtable through embedding. Therefore, as long as a failure can be detected, there is enough information to recover. The difficult part is how to determine that another client has really failed, and the client must follow the fail-stop model. For example, if a client is merely slow but is judged by others to have failed, then that client must stop by itself. Percolator Worker, which is the client mentioned above, uses lease reclaim based on the Chubby service[12] to provide the failure detection and fencing mechanisms described above.

The remaining problem is relatively straightforward. There are two cases based on the state of the primary lock:

The paper does not mention specifically how these affected cells are discovered. Possible methods may include:

Because timestamps must be totally ordered and increasing, the Timestamp Oracle usually uses a centralized design. As a result, it may become a performance bottleneck and requires special optimization.

In Percolator, the Timestamp Oracle uses pre-allocation: it directly allocates a batch of timestamps and persists the largest one to reliable storage. This pre-allocation approach allows allocations for the next period of time to require only memory accesses. If a restart occurs, allocation can simply continue from the largest timestamp that may already have been allocated, read from reliable storage.

To save communication overhead, Percolator Worker obtains a batch of timestamps from the Timestamp Oracle through batching and aggregation.

For a Percolator Transaction, without considering retries and error handling:

This shows that read and write amplification is still quite severe. The performance comparison between Percolator and bare Bigtable is shown in Figure 8 from the paper:

In my view, Percolator’s advantage is that it non-intrusively adds multi-row and multi-table transaction capability to a storage system that supports multi-versioning and row-level transactions. Its disadvantages are: