Conflict Serializability

1.1 Work in Adya's and Fekete's articles

In Adya's paper, except for some insertion statements, every SQL statement is expressed as a predicate read followed by a number of regular reads and writes. The predicate read, which corresponds to the WHERE clause in a SQL statement, is an interpretation of the process of identifying which data objects we are to access. Interpreting such a process as a read is an important conceptual advance, because now we make it clear that a write can affect other SQL statements in two ways: affect the value of a data object that is to be read as usual, or affect whether a data object is selected for access.

A transaction is like a program in a programming language like C. It may have input parameters, return values(for example, in the form of output parameters of a stored procedure that encapsulates the transaction), statements in a regular programming language like branching statements and loops, deterministic math calculations, non-deterministic math calculations(like calling the Rand() function in MySQL) and finally SQL statements. SQL statements are of two types: those that access the database and those that don't. For those that don't access the database, their behavior depends on the environment, just like regular programming language statements; for those that access the database, their behavior also depends on the database state. What is more, no mater what branches the execution might follow, the SQL statements are executed in a sequential order. We can easily find the constructs we've just described in the following

	  
	    The Order-Status transaction(Read-Only):

        if the customer in the order is represented by a name{                         //60% chances

          select count(c_id) into :namecnt from customer 
             where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id;

          declare c_name cursor for 
             select c_balance, c_first, c_middle, c_id from customer
                where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
                order by c_first;

          open c_name;

          if (:namecnt%2) :namecnt++;
          for (n=0;n<:namecnt/2;n++) {
              fetch c_name
                 into :c_balance, :c_first, :c_middle, :c_id;
          }

          close c_name;
        } 
        else if the customer in the order is represented by an id{                    //40% chances

          select c_balance, c_first, c_middle, c_last from customer
             where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;
        }
    
        declare c_order cursor for
           select o_id, o_carrier_id, o_entry_d from orders 
              where o_d_id=:c_d_id and o_w_id=:c_w_id and o_c_id=:c_id
              order by o_id desc;

        open c_order;

        fetch c_order 
           into : o_id, :o_carrier_id, :o_entry_d;

        close c_order;

        select ol_i_id, ol_supply_w_id, ol_quantity, ol_amount, ol_delivery_d from order_line
           where ol_d_id=:c_d_id and ol_w_id=:c_w_id and ol_o_id=:o_id;
      

	                                                                                                       ## 
	  

The next few paragraphs are basically copy and paste of section 4.1 of Adya's paper, which is one of the major contributions to the literature by its authors. If you are familiar with the literature, you may just skim through it.

Database consists of data objects that can be read or written by transactions. While Fekete's paper considers different granularity of objects, e.g., pages, tuples, fields, Adya's paper basically only deals with tuples. Our article will deal with tuples, fields and most importantly, 'decision set's as we'll define.

A data object has one or more versions. However, transactions only interact with the database in terms of data objects; the database system maps each operation on a data object to a specific version of that object. A transaction may read versions created by committed, uncommitted, or even aborted transactions; constraints imposed by some isolation levels will prevent certain types of reads, for example, those created by aborted transactions.

When a transaction writes a data object x, it creates a new version of x. A transaction Ti, i>=0, can modify a data object multiple times; its first modification of data object x is denoted by xi:1, the second by xi:2, and so on and they are called internal versions. Version xi denotes the final modification of x performed by Ti before it commits or aborts. A transaction’s last operation, commit or abort, indicates whether its execution was successful or not; there is at most one commit or abort operation for each transaction. The committed state reflects the modifications of committed transactions. When transaction Ti commits, each version xi created by Ti becomes a part of the committed state and we say that Ti installs xi; the system determines the ordering of xi relative to other committed versions of x. If Ti aborts, xi does not become part of the committed state.

Conceptually, the initial committed state comes into existence as a result of running a special initialization transaction, Tinit. Transaction Tinit creates all data objects that will ever exist in the database; at this point, each data object x has an initial version, called the unborn version. When an application transaction creates a data object x (e.g., by inserting a tuple in a relation), we model it as the creation of a visible version for x. Thus, a transaction that loads the database creates the initial visible versions of the objects being inserted. When a transaction Ti deletes a data object x(e.g., by deleting a tuple from some relation), we model it as the creation of a special dead version, i.e., in this case, xi is a dead version. Thus, data object versions can be of three kinds: unborn, visible, and dead.

If a data object x is deleted from the committed database state and inserted later, we consider the two incarnations of x to be distinct objects. When a transaction Ti performs an insert operation, the system selects a unique data object x that has never been selected for insertion before and Ti creates a visible version of x if it commits.

We assume data object versions exist forever in the committed state to simplify the handling of inserts and deletes, i.e., we simply treat inserts/deletes as write (update) operations. An implementation only needs to maintain visible versions of data objects, and a single-version implementation can maintain just one visible version at a time. Furthermore, application transactions in a real system access only visible versions.

This generic multi-version view of database of Adya's paper is another, yet more important, conceptual advance. Now every concurrency control mechanism can be viewed as a special case of this setup. For example, MySQL InnoDB's MVCC is one that chooses to expose several visible versions of a data object to transactions, while NDB Cluster chooses to expose just one. So if we could prove a generic version of the Serializibility Theorem based on this setup, we could apply it to all the concurrency control mechanisms that satisfy the conditions in the Theorem. This is exactly what we are going to do in this article: killing multiple birds with one rock.

After the concepts of transaction and database model are coined, Adya's paper continues to capture an execution of a database system by a history. A history H over a set of transactions consists of two parts: a partial order of events E that reflects the order of the operations (e.g., read, write, abort, commit) of those transactions, and a version order, <<, which is a total order on committed versions of each data object.

A write operation on data object x by transaction Ti is denoted by wi(xi)(or wi(xi:m)); if it is useful to indicate the value v being written into xi, we use the notation, wi(xi, v). When a transaction Tj reads a version of x that was created by Ti, we denote this as rj(xi) (or rj(xi:m)). If it is useful to indicate the value v being read, we use the notation rj(xi, v). In a committed transaction, wi(xi) writes the committed version xi.

Now I start to list the conditions for Adya's Serializability Theorem.

The partial order of events E of a history satisfies the following conditions:

In a standalone database system like MySQL InnoDB, usually a variant of the temporal order is chosen to be the partial order.

	           t < t' => C(t) < C(t')          Monotonicity Condition(This implies C(t) < C(t') =>  t < t' )                                                                                            
	  

	                                                                                                       ## 
	  

Let's investigate the order of events inside a transaction in a standalone system and how the 'happened before' partial order preserves it. Although there are different execution paths in a transaction due to the loops and conditional branches in it, its execution follows exactly one such path. In this sequential execution of the transaction, SQL statements are executed one by one and the order between events from different SQL statements apparently coincides with the 'happened before' partial order. Also, in SQL statement like 'update t set col=col+1', the reading of col must happen before the writing since this is just the 'read-modify-write' sequence. Again this coincides with the 'happened before' partial order. Some events in a transaction can't be ordered by the 'happened before' partial order, for example, the predicate read and reads/writes that follow. This is because the predicate read is a logical operation and its duration can overlap with reads/writes that follow. But now we have shown that Condition 1 is satisfied with the 'happened before' partial order for standalone database system like MySQL InnoDB and PostgreSQL.

Condition 2 suggests that if a read happens, a write event of the version read exists and precedes the read in the partial order. This is a reasonable requirement if we choose the write event carefully so that a read is only possible after it ends.

Condition 3 is just another reasonable requirement for a transaction processing system and can also be interpreted without the 'happened before' partial order.

Remark: The partial order we just defined uses a local clock satisfying the Monotonicity Condition, which makes it unsuitable for distributed systems like NDB Cluster. We will introduce a new partial order shortly for distributed systems which coincides with the one we just defined when the system happens to consist of one node.

From now on, we'll only consider complete histories. Notice that Condition 4 allows us to only consider committed transactions since atomicity guarantees the aborted ones NOT to have any effect on the system. So we'll only consider systems that can provide atomicity from now on.

For a standalone database system like MySQL InnoDB or PostgreSQL, the total order << we impose on the committed versions of a data object is just the temporal order of the write events. For distributed system like NDB Cluster where versions of the same data object might migrate from one machine to another(for example, when NDB Cluster performs a re-partition if a new node is added). In this case, an unambiguous total order can still be defined if we bring the order of machines in the migration path into the picture.

Now let's take a closer look at predicate read, a concept closely related to the process of identifying which data objects are to access, which is a key component of Adya's paper. A predicate read corresponds to the predicate in the WHERE clause in a SQL statement, which we use to decide what tuples we'll return for access in the statement. In some cases, the predicate read is exactly the process of identifying which data objects are to access. This is, for example, the case when we use a unique index to access the database, like the following statement in Example 1 when we we access a customer's info with an id:

      select c_balance, c_first, c_middle, c_last from customer
         where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;                                                                                                
	

In other cases, the predicate read is only part of the process of identifying which data objects are to access. This is, for example, the case when we use a non-unique index to access the database, but only part of the returned tuples qualifies. For instance, in Example 1, when we access a customer's info with the last name, we first return all the customer with that last name and then filter out the mid-point one in a specific order in a cursor to access. This is demonstrated with the following statements:

      declare c_name cursor for 
         select c_balance, c_first, c_middle, c_id from customer
            where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
            order by c_first;

      open c_name;

      if (:namecnt%2) :namecnt++;
      for (n=0;n<:namecnt/2;n++) {
          fetch c_name
             into :c_balance, :c_first, :c_middle, :c_id;
      }

      close c_name;                                                                                    
	

In this case, the process of identifying which data objects are to access equals the predicate read specified in the WHERE clause, plus the cursor iteration. For now, we will only consider statements referencing only one table. Statements with complex constructs like joins, sub-queries or unions will be handled later.

As the name suggests, we consider predicate read as a type of read. Hence a write can affect its result. A new pair of conflicts arise from this: Directly predicate-read-depends and Directly predicate-anti-depends. Now let's see how Adya's paper define these two conflicts. Please pay close attention to them because it is these definitions that lead to Example 0.

Notice the versions in Vset(P) could be either committed or internal ones. Predicate P is then evaluated against version set Vset(P), the tuples with a version in the version set that satisfies the predicate are chosen for later access. This evaluation process is the realization of a predicate read. A predicate read performed by transaction Ti is denoted by ri(P: Vset(P)) and let's call this subset of Vset(P) that satisfies the predicate the matching set. A write can change the matching set in two ways: it changes a non-matching version outside the set into a matching version inside the set(the IN operation) or it changes a matching version inside the set into a non-matching version outside the set(the OUT operation). Adya's paper define this formally with the following:

The definition above identifies Ti to be a transaction where a change occurs for the matching set of rj(P: Vset(P)). Note that i can be equal to j.

Notice that in Definitions 3 and 4, the write that conflicts with the predicate read doesn't have to install a version that is in Vset(P) or immediately after the one in Vset(P) .

        update t set id3 = 5 where id1 = 1;

        update t set id3 = 4 where id1 = 1;
	  

        select value from t where id2=2 and id3=5;
	  

	                                                                                                       ## 
	  

Definitions 1-4 are the way that Adya's paper tries to capture how a write affects a predicate read and it makes sense because everything there is tuple based. But Fekete's paper aims to generalize Adya's theory to other granularity, including field granularity and it seems they think that is just a process of copy and paste. Let's re-examine Definitions 1-4 and see if that is the case. In the next few paragraphs, we'll only pay attention to field granularity, since we hope it will help to minimize conflicts.

Remark: From this point on till the analysis of Example 0, all about Fekete's paper are my understanding of how they generalize Adya's work since some of the aspects require explanation which they didn't provide. My work, which is inspired by Adya's and Fekete's, starts after the analysis of Example 0. Whether my understanding about Fekete's generalization is correct or not doesn't affect the correctness of my work, in which every definition is in math and every major theorem is proved. However, if my understanding about Fekete's generalization is correct, the correctness of PostgreSQL's serializability implementation is in doubt because the existence of Example 0. Even if my guesses are wrong, they still should prove a version of the serializability Theorem since the one Adya's work relies on may not be appropriate for a modern workload as we'll see.

For Definition 1, Vset(P) just consists of field versions and these field versions are generated by field writes. And the matching set ri(P: Vset(P)) is the set of tuples where the corresponding field versions in Vset(P) satisfy the predicate P.

For Definition 2, it makes sense for the case when the predicate read involves only one field. It doesn't when it involves more than one field and this will become clear when we define what a 'decision set' is.

For Definitions 3 & 4, I am quoting Fekete's paper as follows:

"Definition 2.1 (Transactional Dependencies). Following the concepts of Adya et al. [2000], but with new terminology, we characterize several types of transactional dependency in an interleaved SI history.

We say that there is a Tm → Tn dependency (a predicate-write-read dependency) if Tm installs a data item version Xm so as to alter the set of items retrieved for a predicate read by Tn, and also the commit of Tm precedes the start of Tn, so that the result of the predicate read by Tn takes into account a version of data item X equal to or later than the one installed by Tm.

We say that there is a Tm → Tn dependency (a predicate-read-write dependency or predicate-anti-dependency) if Tn changes a data item X to version Xn so as to alter the set of items retrieved for a predicate read by Tm, and also the commit of Tn follows the start of Tm, so that the result of the predicate read by Tm takes into account a version of data item X prior to the one installed by Tn."

As contrary to Adya's paper, these definitions are defined purely in English, instead of math. So I have to guess what they actually mean. From the context, these are just restatement of those in Adya's, only this time for Snapshot Isolation. So 'so as to alter' is just 'change the matches of a predicate read ', in particular when a 'data item' is of field granularity. Again if the predicate in concerned is relevant to exactly one field, there is no ambiguity about this statement. But what if it is relevant to multiple fields like the last update in T1 of Example 0? What exactly does it mean by saying 'a field write to alter the set of items retrieved for a predicate read'?

Let's continue examining other conditions for Adya's Serializability Theorem.

These three definitions define the common conflict types in consistency theory. Notice that the data object versions in these three definitions are of tuple level granularity in Adya's paper. But they are of any granularity, including field, tuple, page level in Fekete's paper.

We sometimes also combine Definitions 3 & 5, 4 & 6 and call them Directly Read-Depends(WR conflict), Directly Anti-Depends(RW conflict) respectively for convenience.

Finally we are at a place to present the conditions relevant to conflicts that the serializability isolation level in Adya's paper has to satisfy. For a history H to be serializable, the following three conditions must be proscribed.

        w1(x1:i) : : : r2(x1:i) : : : (a1 and c2 in any order)

        w1(x1:i) : : : r2(P: x1:i, ...) : : : (a1 and c2 in any order)
                                                                                        
	  

        w1(x1:i) : : : r2(x1:i) : : : w1(x1:j ) : : : c2

        w1(x1:i) : : : r2(P: x1:i; :::) : : : w1(x1:j ) : : : c2
	  

Proscribing Condition 6 means the framework is essentially based on a Read-Committed isolation level. This assumption, however, is already implied in the definitions for different kinds of conflicts since versions involved are all committed ones. We already know that versions in Vset(P) could be either committed or internal ones, but reading of internal ones only happen when the corresponding write is in the same transaction and hence doesn't constitute a conflict.

Everything between Definition 5 and Condition 7 applies to data object of field granularity and hence is ready for Fekete's generalization, baring the issue with predicate read with multiple fields we've mentioned in Definition 2.

	                                                                                            ( To be continued … )                   
	  

1.2 Building on top of Adya's and Fekete's work

Example 0(the NDB Cluster part) also indicates if we were to develop a field level Serializability Theorem for system like NDB Cluster, which is a major goal of this article, we must also pay attention to similar problems we've just described. Fortunately, Adya's paper has set up a framework that makes everything easy. We've introduced their work along the way. What we need is to fill in a few blanks and a suitable version of serializibility theorem will pop up. So let's get to it.

First let's explain why a serializability implementation based on data objects of field level granularity is important for 'pessimistic technology'. In 1993, right after SQL-92 was released, Alexander Thomasian published the paper [To 93], where thrashing behavior is shown when a system with two phase locking is under loads which cause too many locks. From that point on, when people ask why 'pessimistic technology' like MySQL InnoDB doesn't use their two phase locking based serializable isolation level as the default isolation level, the answer is usually 'its performance degrades under loads with excessive locks' and the like. But if Thomasian's paper is examined more carefully, one probably can see what really causes the thrashing behavior is 'excessive lock waits'.

If the real reason behind this thrashing behavior is 'excessive lock waits', there are two obvious approaches to reduce them: reduce the lock waiting time and reduce the number of conflicting locks. NDB Cluster has done the first one for us by moving all data from disk to memory. Consider the example of a locking read: for a disk based database system, the time of holding this read lock depends on whether the data object is in the main memory or on the disk. When the read is a miss, any transaction waiting on this lock to be released just have to wait for the data object to be fetched from the disk. NDB Cluster certainly minimizes the wait time due to disk accesses.

For minimizing the number of conflicting locks, consider the following

        select id2 from t where id1 = 1;

        update t set id3 = 4 where id1 = 1;                                                                                               
	  

	                                                                                            ##                   
	  

After we understand the importance of developing a serializability implementation that is based on field level granularity data objects, it is time to define a field version system and impose a total order on it. Today's commercial systems like NDB Cluster, MySQL InnoDB, PostgreSQL still commit their modifications at the tuple level and derived committed field versions can be defined by splitting the committed tuple write into several committed field writes(only for the fields that are modified). And a tuple version can be viewed as an array of field versions. The following example demonstrates this idea.

        update t set value = 1 where id1 = 1;

        update t set id2 = 3 where id1 = 1;

        update t set id3 = 5 where id1 = 1;

        update t set value = 2 where id1 = 1;          
	  

	                                                                                            ( To be continued … )                   
	  

Earlier we've shown that we may impose a total order on the committed tuple versions even if the tuple migrates to another node(like the re-partition NDB Cluster performs when a new node is added). So we can define a total order on the committed field versions as follows: a committed field version is before another if the underlying tuple version is so. It is easy to prove that this is a total order and we call it the 'derived order' on committed field versions. Of course, if we want to build a system that is completely free of tuple versions, we need to assume that a total order on the field versions can be imposed. For instance, in a system with a field level capable locking system, we may use the order of locking events on a field to introduce such an order.

Besides the committed versions, we also need to define an order for the internal versions of a data object inside a transaction in the case it is modified more than once. We've explained earlier before Example 1 that the execution of a transaction is a sequential process, namely, on a statement by statement basis. So assuming each statement doesn't update a data object twice, the order of these versions is obvious, for both tuple and field level versions.

Now let's develop partial order that is suitable for a distributed system like NDB Cluster and also an extension to that defined in Example 2.

        t < t' => Ci(t) < Ci(t'), for all 0 < i < m+1                                                                                             
	  

	                                                                                            ##                   
	  

Before we prove that Conditions 1, 2, 3 & 4 with the 'happened before' partial order are satisfied by MySQL InnoDB, NDB Cluster and PostgreSQL, let's alter Conditions 2 & 3 a bit so that they are more appropriate for our Serializability Theorem. Notice that Example 0 is not affected by this change, so it is OK we didn't do this earlier.

In Condition 2' we didn't claim wi(xi:m) 'happened before' rj(P: Vset(P)). A predicate read could be a lengthy process, for example a full table scan in some cases, and wi(xi:m) could overlap with rj(P: Vset(P)).

Here the granularity of x may be of field level or tuple level for now. We will also consider a 'decision set' level when we define it.

For MySQL InnoDB and PostgreSQL, the 'happened before' partial order is just the one defined in Example 2. And we've already proved it satisfies Condition 1 & Condition 2' in an item read case. The argument for a predicate read case for Condition 2' is similar. For NDB Cluster, a proof need to be delayed until next section after we explain its commit logic. So we are good with Conditions 1 & 2'. Condition 3' is just another reasonable requirement for transaction processing system and can be interpreted without the 'happened before' partial order. We will also require the transactions to be complete, hence Condition 4 is satisfied too.

We earlier explained a total order can be imposed on the committed versions not only in standalone system like MySQL InnoDB and PostgreSQL where the write events of the same data object happen in a serial order at the same node, but also in the case where NDB Cluster performs a re-partition if a new node is added. In general, we may impose a stronger condition on the write events of the same data object by requiring that they happen in a serial order so that it coincides with the 'happened before' partial order. Notice that not every pair of temporally isolated events are related by the 'happened before' partial order, consider events in two nodes which don't communicate with each other for an example. This stronger condition implies the version order of the committed versions coincides with the 'happened before' partial order too, namely, one version is before another if and only if it 'happened before' another. From the definition of the 'happened before' partial order we can easily see that this stronger condition again applies to all three implementations in concern: MySQL InnoDB, PostgreSQL and NDB Cluster, for both tuple versions and derived field versions.

Not only the mathematical representation of the temporal order need to be re-defined, but also the definition of a serial history. A serial history in a standalone system like MySQL InnoDB is usually defined as a sequence of transactions executed 'one by one', which implies the first event in a transaction happens after the last event of the transaction immediately before it if such a transaction exists. This doesn't make too much sense in a distributed system like NDB Cluster because the first event of a following transaction may not happen in the same node as the last event of the previous transaction does and this issue must be addressed in the new definition. Also one consequence of a serial history is that a transaction sees the latest modifications made by previous transactions, but not the ones after it. We need to define a serial history in a distributed system so that this property holds.

Remark: In NDB Cluster, what we have to do is open a mysql prompt, execute the ordered transactions one by one and only start a new one after the previous one commits. In this case, the MySQL node that the prompt connects to serves as the coordinator. So the coordinator here is very different from a transaction coordinator in NDB Cluster.

Notice the requirement that a transaction in a serial execution to always read the latest available committed version of data object is in general NOT appropriate for histories where transactions may overlap. We can easily find such examples in Snapshot Isolation or Repeatable-Read isolation levels where a read reads an older version of an object when multiple versions are stored in the database. And in NDB Cluster's Read-Committed isolation level, MySQL InnoDB's Read-Committed and Repeatable-Read isolation levels, PostgreSQL's Snapshot Isolation, we can always arrange the transactions to be isolated enough so that this requirement is satisfied.

From this definition, it is clear that the order of transactions coincides with the order of write events of the same objects in the 'happened before' partial order, hence coincides with the total order of the object versions.

We assume that when a transaction reports back to the API coordinator its state as committed, all of its modifications are available for reading. This assumption is reasonable since distributed systems usually use two phase commit as their commit logic; in two phase commit a local component commits its modifications, hence ready for reading, before it sends the committed state back to the coordinator. For example, this assumption is satisfied by NDB Cluster and one can see that when we explain the commit logic of NDB Cluster in next section.

With this assumption, it's easy to see that a later transaction in the execution sequence sees the latest modifications in transactions before it: the time for the modifications of a previous transaction to be available is before the time for the commit message to reach the coordinator, which is before the time to start a later transaction, which is once again before the time the reading happens; and since the transaction reads the latest available modification, it reads the latest modifications in transactions before it.

On the other hand, an earlier transaction in the execution sequence sees no modification of a later transaction. Suppose the opposite was true and an earlier transaction could read a modification of a later transaction. The write operation 'happened before' the read operation for any reasonable implementation such that Condition 2' is satisfied with the 'happened before' partial order. However, the read operation 'happened before' the commit of the earlier transaction, which 'happened before' the start of the later transaction, which again 'happened before' the write operation. This contradicts with the fact that 'happened before' is a strict partial order.

Remark: Adya's Serializability Theorem assumes the partial order to be a generic one satisfying Conditions 1, 2, 3 & 4. It is very difficult to prove a generic version of Theorem like that. Of course, the fact that we can't prove it doesn't mean it is not provable. But we will only prove one in which the partial order is 'happened before'.

Definition 9 is also applies to standalone system like MySQL InnoDB where the coordinator and the only local component are co-located together. In that case it becomes the regular serial history definition where the transactions are executed one by one. Also this serial history concept can easily be generalized to a set of SQL statement snippets so that each entity in the set is just a subset of a transaction.

Next let's redefine the conflict types. Definitions 5, 6 & 7 remain the same, only this time field level data objects are included. In the Serializability Theorem we are going to present shortly, Conditions 5 & 6 will be proscribed. This makes the isolation level to start with practically become Read-Committed, so the object versions in Definitions 5, 6 & 7 are all committed versions.

Finally we are at a place to redefine the predicate-based conflicts that might have caused the appearance of Example 0. Starting from this point, we will constrain the predicate in a SQL statement to be one that only involves one table and complex predicates involving more than one table, like joins, sub-queries, unions, will be handled later. We define the set of columns in the only table that a predicate P uses to decide whether a tuple is selected for later access to be the 'decision set'. We also define the set of columns a write(a field write or a tuple write) changes to be the write set of that write operation.

        update t set id2 = 2, id3 = 3 where id1 = 1;

        select * from t where id1 = 1 and id2 = 2 and id3 = 4;                                                                                                      
	  

	                                                                                            ##                   
	  

The Serializability Theorem that we are going to present next comes in three flavors with varying conditions on data object granularity. We will also redefine Definitions 1-4 so that they are suitable for the various flavors of it.

Type A: All the conflicts are based on tuple versions. It is used to demonstrate that Adya's Serializability Theorem can actually be proved, without resorting to the results in [Be 87].

For this type, the first internal tuple version generated by a write operation need to be paid more attention to. Besides the fields that are actually written in the write operation, there might be fields that are not. Since we are generating a tuple version, we have to fill in a value for each of those. Where do these values come from?

There are three cases. In the first case, all fields in the tuple are modified and we need to do nothing. In the second case, a tuple read 'happened before' the write in the same statement like in 'update...set col=col+1'; in this case, the values naturally come from the tuple read. In the third case, it is a blind write like 'update table grades set grade=”A”'. In this article, we assume the following:

The third case works the same as the second one except that it reads a tuple version implicitly before the write. In other words, there is no real blind write in our framework.

For MySQL InnoDB and NDB Cluster, the tuple is locked before it is updated and the locked tuple's value serves as the implicit read; for PostgreSQL, the write is always against a tuple version in the snapshot taken when the transaction starts and that tuple version serves the same purpose. This is so since no other transaction can write the tuple before the designated one does in any of the three cases. So the assumption works for the three implementations concerned in this article.

For predicate-based conflicts, Definitions 1 & 2 remain the same as in Adya's paper. The digression starts with the added definition for the 'conflict with' concept. The item-based conflicts remain the same as in Adya's paper, of course.

Type B: Updates in transactions still generate tuple versions, both internal and committed ones. Field versions are derived from these tuple versions as usual. All the regular item-based conflicts are interpreted at field level, namely, based on the derived field versions.

Like the derived field versions, we also derive 'decision set' versions from tuple versions. Namely, a derived 'decision set' version is generated whenever fields in the 'decision set' are modified in a tuple write and these derived 'decision set' versions are ordered the same way as their associated tuple versions.

	                                                                                            ##                   
	  

For predicate reads, we use derived 'decision set' versions to capture the conflicts. For Definition 1, Vset(P) consists of a set of derived 'decision set' versions for the tuples to be matched. For Definition 2, we also use derived 'decision set' versions to replace the tuple versions in it. In other words, the versions xi, xh in the following restated definition are derived 'decision set' versions.

And we'll use derived 'decision set' versions to capture predicate-based conflicts as we'll describe in Definitions 3' & 4' shortly.

There is a new kind of conflict in this type of the Serializability Theorem though. They are the WR, RW and WW conflicts between reads and writes of the committed derived 'decision set' versions, just defined like those for tuple or field versions. It is easy to see that the serial order of write events on the 'decision set' coincides with the 'happened before' partial order since that of the tuple write events do. After the introduction of this new set of conflicts, we also include them into the item-based conflicts, as opposed to the predicate-based ones. In particular, Definitions 5, 6 & 7 for type B include committed 'decision set' versions now, not just committed field versions.

As in the tuple write case in type A, we may require an implicit 'decision set' version to be read before a 'decision set' version write if not all the fields in the 'decision set' are written. That is because in a system that still generates tuple versions, a 'decision set' version write where not all the fields in the 'decision set' are written implies a tuple version write where not all the fields in it are written. Hence we may require that a tuple read must precede such a tuple write as in the type A case and the derived 'decision set' read will serve our purpose. Similar to the type A case, MySQL InnoDB, NDB Cluster and PostgreSQL all fulfill this requirement. This implicit 'decision set' version is called the initial internal 'decision set' version. And this read is the only case that a 'decision set' read is referred to as an item read, all the others are involved in a predicate read. Notice this implies that in MySQL InnoDB, NDB Cluster and PostgreSQL, item RW conflict for 'decision set' actually doesn't exist since the next version is written by the accompanying write, not by another transaction. In MySQL InnoDB and NDB Cluster, that is because the read happens after the lock for the accompanying write is acquired and this lock will block the write from another transaction until the accompanying write finishes; in PostgreSQL, the two writes on the 'decision set' are considered to be writes on the same tuple and their associated transactions will be non-concurrent under Snapshot Isolation and hence the conclusion.

For possible item reads following a predicate read, we will use the 'derived' field versions from the same tuple version associated with the matching 'decision set' version.

Remark: The recognition of conflicts between 'decision set' versions is a key abstraction for the development of type D of the Serializability Theorem, which is free of tuple versions. When we apply type B of Serializability Theorem to NDB Cluster, we will see how these conflicts are buried under regular tuple lock conflicts.

This type of the Serializability Theorem is designed for the current generation of technology where data objects are mainly accessed by tuple. Notice that in the proof we will try to conclude that the field and 'decision set' versions, NOT the tuple versions, are read and written in the same way in a given history and its serial peer. An example will be given to demonstrate this after the proof.

Type C: This type of the Serializability Theorem still requires the tuple concept which corresponds to the row concept in a table, but it doesn't require the concept of tuple versions. It does require the existence of field versions and a total order generated by serial write operations on that field that is consistent with the 'happened before' partial order. All conflicts are interpreted at the field level and this implies that the 'decision set' of a predicate read only consists of one field. It also requires a way to identify which field version to read after a predicate read if item reads follow.

Type C of the Serializability Theorem is designed for the future, for example, when a field level capable locking system is available. Notice that the item-based conflicts between reads and writes of 'decision set' versions are just those between field versions.

In type C, the requirement of the way to specify what field versions to read after a predicate read is matched is implicit: as long as it makes a serial history satisfy its key property, namely, when a transaction is executed, it always reads the latest committed version of data object that is available.

So Definitions 1 & 2 are syntactically identical for all three types of the theorem with the item versions in concern being of different granularity. Before we present the Serializability Theorem, we need to redefine Definition 3 & 4. Let's start with the following

Version xi in this definition could represent either a field, a tuple or a 'decision set' version respectively in type A, B & C of the Serializability Theorem.

Note that in Adya's paper, it doesn't emphasize the write in Condition B that changes the match to be the closest one, although in general there could be more than one write that does so as the following example indicates.

        T1: update t set id1=2 where id1 = 1

        T2: update t set id1=1 where id1 = 2

        T3: select * from t where id1 = 1                                                                                                  
	  

	                                                                                            ##                   
	  

Now let's re-define Definitions 3 & 4 to suit our needs.

Again a version like xi in these two definitions represents a tuple, a 'decision set' and a field version in type A, B & C of the Serializability Theorem respectively. Notice there is an extra requirement that Tj doesn't generate the version after xk itself in Definition 4' comparing with Definition 4 in Adya's work. If Tj does generate the version after xk itself, the dependence between Tj and Ti is captured with a sequence of WW conflicts on x.

Remark 1: We claim that Tj in Definition 4' generates the version after xk when it also writes x. In Adya's and also this article's framework, predicate read and the following reads/writes are interpreted as different operations. This might seem to imply that it was possible to insert a write operation by another transaction between them at first glance. For example, it might seem that after the predicate read identified a tuple as a matching one, another transaction could update it before the original transaction performed a read. But actually in today's page-oriented implementation of database system, this irrational behavior is impossible. For instance, in a lock based system like MySQL InnoDB, typically before a tuple is identified as matching, its containing page has to be locked with a semaphore; after it is identified, a lock on the tuple will be requested. In general there is no room for another transaction to intervene with this process with a write, even in the more complex case where the requested lock can't be acquired immediately and a lock wait becomes a must, and then the semaphore need to be given up and re-acquired until the lock is obtained to check if the path structure to the page has changed since then. In a lock free system like Snapshot Isolation, such behavior is impossible either since it would introduce conflicting updates in two concurrent transactions. So this claim is sound.

Remark2 : Such a modification is necessary for Definition 4' since Tj's write on x might change the match, but it doesn't constitute a conflict since a conflict is between two different transactions. Nevertheless, it does make xi lose the status of being closest to xk and violates Condition B. Notice in Adya's work, it doesn't require xi to be closest to xk and hence doesn't need this extra requirement. Such a modification is not necessary for Definition 3' since no committed version is generated before the predicate read in Tj.

We also need to re-visit Conditions 2' & 3' here. Although this time in the predicate case, a version of x doesn't just refer to tuple or field version, it also refers to 'decision set' version.

The fact that MySQL InnoDB and PostgreSQL satisfy Condition 2' in the 'decision set' case is clear since the 'decision set' versions are derived from tuple versions and Condition 2' is satisfied in the case of tuple versions. The fact that NDB Cluster satisfies in the 'decision set' case Condition 2' will be delayed after we explain its commit logic. And Condition 3' is again just a reasonable requirement for the 'decision set' case.

We also need to re-visit Definition 8 since it also involves 'decision set' versions.

Here we combine Definitions 3' & 5, 4' & 6 and call them Directly Read-Depends(WR conflict), Directly Anti-Depends(RW conflict) respectively for convenience.

Definition 8 remains syntactically identical although predicate-based conflicts are semantically different for the three different types. And for type B, we also have the item-based conflicts between reads & writes of derived 'decision set' versions.

Now we are ready to define a serializable history.

Remark: One can generalize this concept of equivalence to a subset of statements in history H and H'. Also it is easy to prove that the same serial execution of transactions with different coordinator to be equivalent to each other and the definition of serial history we defined earlier is sound.

1.3 Type A, B, C of the Serializability Theorem and the proof

*****************************************************************************************************************************************               
	  

This theorem and its proof is the core of this project and the correctness of my work largely depends on it. You are invited to review it
for me to make sure it is absolutely flawless. The proof is lengthy(about 10 pages), but startforward(nested induction of two levels).
Anyone with a decent math education should be able to comprehend it. For geeks like us, it could entail a fun weekend. Your help is
highly appreciated.              
	  

*****************************************************************************************************************************************               
	  

	                                                                                                       ## 
	  

1.4 An issue with Condition 3'

Condition 3' basically says inside a transaction, a read will see writes to the same item written by itself. This may not be true for some systems, particularly those implementing Snapshot Isolation, which will read only from a snapshot identified by a transaction's starting timestamp. In that case, we must prove a Serilizability Theorem that a read will NOT see writes to the same item written by itself. But that is a trivial task since in the proof of such a theorem, we only handle the case this kind of writes never happen and remove all the arguments dealing with the case when they do.

1.5 Neighborhood of match change

In the process of proving the predicate reads in H and Hs give rise to same set of tuples in all three types of Serializability Theorem, we know that the predicate reads in H and Hs may not use the same version to make the match decision, but the difference between them can't contain one that changes the match of the predicate. So if we define the following

Then in H and Hs, corresponding predicate reads use versions from the same neighborhood of match change to make decisions for every tuple.

1.6 Comparing three types of the Serializability Theorem

        T1: select value from t where id1 = 2;

        T2: update t set id2=3 where id1 = 1;

        T3: update t set id3=5 where id1 = 1;
	  

	                                                                                                       ## 
	  

	                                                                                                       ## 
	  

1.7 Generalized Serializability Theorems

Historically, there are different flavors of serializability other than the one in the Serializability Theorem we've just proved, namely, conflict serializability. One of them assumes that there are no conflict loops among the Read-Write transactions in the history. In this case, even if there is a loop in the history, there still exists a serial history of the Read-Write transactions that will leave the database as in the same state as the original history does.

	                                                                                                       ## 
	  

Remark: There is actually an issue about the statement 'the fact that a Read-Only transaction doesn't write the database means that it can't affect the behavior of the Read-Write transactions' in this proof. A call to the RAND() function in a Read-Only transaction can affect a following Read-Write transaction in a very subtle way: the value u returned by RAND() in a Read-Only transaction is from a sequence that simulates the random behavior of RAND(); so if a Read-Write transaction later calls RAND(), the returned value v has to be one after u in that sequence. Think about the situation in one history a Read-Only transaction calls RAND() while in the other history the corresponding Read-Only transaction doesn't since they don't have to behave the same now; a subsequent Read-Write transaction in both histories will call RAND() and they return different values. However, this can be handled with a similar argument as for the RAND() case in the Serializability Theorem: C is equivalent to a serial execution of itself in which corresponding calls to RAND() returned the same values; or alternatively, C can be viewed as an execution with returned values from RAND() being constants. So it doesn't pose as an actual problem.

                    T1                                                                             T2

        start transaction;

        select value from t where id1 = 1;

                                                                                   start transaction;

                                                                                   select value from t where id1 = 1;

        update t set value=value-100 where id1 = 1;

        select value from t where id1 = 2;

        update t set value=value+100 where id1 = 2;

        commit;

                                                                                   select value from t where id1 = 2;

                                                                                   print the sum of the values from the last two                             
                                                                                   statements to the screen for the customer

                                                                                   commit;                                                                                  
	  

	                                                                                                       ## 
	  

What is demonstrated in this example is summarized in the following

	                                                                                                       ## 
	  

Generalized Serializability Theorem I suggests that if we could device a way to make sure the Read-Write transactions do not contain a conflict loop, we may only see inconsistencies through the Read-Only transactions. What if we only want to see inconsistency from some, but not all of the Read-Only transactions? The following theorem provides an answer.

	                                                                                                       ## 
	  

This is not a surprise either since this time we exclude a possibly smaller set of transactions that can't affect database writes than in Generalized Serializability Theorem I. And the following corollary is true for Generalized Serializability Theorem II:

	                                                                                                       ## 
	  

We'll see how these theorems are applied to the TPCC benchmark example in next section.

Now let's apply the Serializability Theorem to the three SQL implementations—PostgreSQL, MySQL InnoDB and NDB Cluster.

In particular, this could provide a fix to PostgreSQL's serializability implementation if Fekete's generalization did cause a problem. On the other hand, Cahill's paper([Ca 08]) doesn't provide any theoretical change to [Fe 05], so as long as we fix Fekete's possible problem, PostgreSQL's serializability implementation will be sound again if it is not already so.

MySQL InnoDB already comes with a serializability implementation, which is based on Two Phase Locking, or 2PL. One can actually use the Serializability Theorem we've proved to show that 2PL is really serializable(a short proof is given in Appendix B). But even then we may still do better for some application with our Serializability Theorem. That is because MySQL InnoDB's serializability implementation is of tuple level granularity and field level granularity implementation based on type B of the Serializability Theorem will have advantages over it for some workloads. However since MySQL InnoDB is not the focus of this article, I will leave it for the interested audience and they should be able to get some hint about how to apply it from next section when we apply the Serializability Theorem to NDB Cluster.

We will fill in the blanks for Corollary 3 in next section.

So we've proved three versions of the Serializability Theorem for three SQL implementations, PostgreSQL, MySQL InnoDB and NDB Cluster, as we promised , thanks to the powerful framework set up by Adya's paper. If this was another research paper that is trying to prove a version of the Serializability Theorem, we already did it, for three times this time. But what we are looking for is a practical solution to the serializability problem that is useful for many, if not most, of the sophisticated database application developers. To this end, we are just at square one and a lot of work and testing are needed for that purpose. So in next section, we will device a way to apply these theorems to NDB Cluster so that histories executed within it will be serializable and start the journey. Join me in the quest when you feel ready!

2.1 Commit logic of NDB Cluster

The commit mechanism in NDB Cluster is a variant of two phase commit, which is a commit protocol commonly used in distributed transaction processing system. In the two phase commit protocol, data reads and writes in a transaction happen in different participating nodes across the system and one node is chosen as the transaction coordinator(TC) to orchestrate this process. The transaction coordinator starts the two phase commit protocol after all SQL statements are executed and the transaction is ready to commit. The protocol consists of two phases as the name suggests. In the first phase(voting phase), a message like 'are you ready to commit?' is sent out to each and every participant by the coordinator. When all replies are received by the coordinator, a 'commit' consensus is reached if all answers are affirmative, otherwise an 'abort' decision is made. This is the end of the first phase and the second phase starts right away. In the second phase, the coordinator sends out messages like 'commit now' or 'abort now' to all participants based on the decision made in the first phase. Upon receiving the message, participants perform operations necessary for the commit(the so called local commit) or abort request. After that, a message is sent back to the coordinator reporting this. Upon receiving all the reporting messages from the participants, the coordinator reports back to the system that the transaction is committed or aborted and ends the execution of the two phase commit protocol.

In NDB Cluster, the transaction coordinator role is assumed by a transaction coordinator block(also called TC) in the NDB kernel. A transaction consists of three phases: prepare phase, commit phase and complete phase. The voting phase and part of the second phase of the two phase commit, specifically the part of operations relevant to an abort decision, happen in the the prepare phase. And the part of the second phase in the two phase commit that is relevant to a commit decision is split into the commit phase and complete phase in NDB Cluster. In other words, at the very end of the prepare phase, TC verifies if the transaction is ready to commit. If the answer is 'no', it will abort the transaction and the later two phases will never happen. If the answer is 'yes', the commit phase starts and is followed by the complete phase.

In the commit phase, TC sends out a commit request to each participant. Upon receiving the request, a participant starts its local commit process. If a tuple is updated in this participant, the write lock on the primary replica will be released. The new tuple version will be available both for non-locking read as in the Read-Committed isolation level and locking read as in 'select … lock in shared mode' or 'select … for update' right after the release of the lock at the primary replica. However, the write lock on the secondary replica is not released until the complete phase. After all the operations necessary for the local commit are performed, a confirmation message is sent back to TC to signify the ending of the local commit. After all the confirmation messages are received by TC, it completes the commit phase by sending a commit acknowledgment(reporting its state as committed) to the API node that initiates this transaction.

Then the complete phase starts with TC sending out a complete request to each participant. Upon receiving the request, a participant starts its local complete process. Among other operations, the write lock on the secondary replica will be released. After all the operations necessary for the local complete are performed, a confirmation message is sent back to TC to signify the ending of the local complete. After all the confirmation messages are received by TC, the complete phase ends.

2.2 Completing the proof of the Serializability Theorem for NDB Cluster

Now it is time to fill in the blanks we left out in the last section about the 'happened before' partial order and finish the proof of the Serializability Theorem.

First, the following is assumed in the discussion of a serial history in the last section: in the discussion of a serial history, when a transaction reports back to the API node its state as committed, all of its modifications are available for reading. This is obvious from the commit logic of NDB Cluster.

Next, let's show that the partial order 'happened before' satisfies Conditions 1 & 2'.

In the Condition 1 case, for events happened in TC that are ordered in the transaction, this order is preserved in the 'happened before' partial order by condition 1 of Example 6. For instance, the 'start of transaction' event 'happened before' the 'start of commit phase' event, which 'happened before' the 'end of commit phase' event, which alternatively 'happened before' the 'start of complete phase' event and so on. On the other hand, let e and e' be events in different SQL statements S and S' respectively so that S is executed before S' in the sequential execution of the transaction. For example, e could be a write of a certain field and e' could be a read of another field. Events e and e' could happen in different nodes in general. But e 'happened before' the ending event of S since a message is sent back to TC notifying the write of the field, and the ending event of S 'happened before' the starting event of S' because both events happen in TC, which in turn 'happened before' event e' since a message is sent from TC to another node to request the read. Also in a SQL statement like “update t set col=col+1” where the read of col 'happened before' the write of it can be argued similarly. This means NDB Cluster satisfies Condition 1.

Remark: In a distributed system case like NDB Cluster, the item reads/writes after a predicate read might overlap with each other in Condition 1 since they are usually executed by different threads on different nodes.

In the Condition 2' case, the fact that creation of a data object version 'happened before' the reading of it is transparent from the commit logic of NDB Cluster even if a re-partition exists like in the case a new node is added to the cluster.

We've already explained Conditions 3' & 4 are both satisfied, while Conditions 5 & 6 are proscribed by NDB Cluster.

Remark: Condition 1 is actually not used in the proof of the Serializability Theorem, but it is heavily depended on for the rest of this section. I keep it as a condition for the Serializability Theorem so that it is easy to compare it with Adya's work. It might be moved to a place after the proof, like here.

So as long as we can device a way to enable histories executed in NDB Cluster to be free of conflict loops, we may apply the Serializability Theorem to NDB Cluster and achieve serializability. At first glance, it might seem we may apply both type A and type B of the Serializability Theorem to NDB Cluster. But the following example suggests it might not be appropriate to apply type A to it.

                     T1                                                                T2

        start transaction;

        update t set value=0 where id1 = 1;

                                                                                start transaction;

                                                                                select id1 from t where id2 = 3 lock in share mode;

	  

	                                                                                                       ## 
	  

Remark: This example also applies to MySQL InnoDB since it uses a similar locking mechanism when it comes to access data with indices. It doesn't apply to Snapshot Isolation of PostgreSQL though since it doesn't use locks to control access.

2.3 Application of the Serializability Theorem to NDB Cluster

Before developing a systematic way to achieve serializability in NDB Cluster, we'd like to use an example to show that the Serializability Theorem is already useful.

                                 T1                                T2                                          T3 

        start transaction;

        select a tuple with primary key k;

                                                            start transaction;

                                                            delete the tuple in T1;

                                                            commit;

                                                                                                        start transaction;

                                                                                                        insert a tuple with the key k;

                                                                                                        commit;

        select a tuple with primary key k;

        commit;
	  

	                                                                                                       ## 
	  

For more complex applications in NDB Cluster, conflict loops are possible and we will device a systematic way to eliminate them. The following theorem suggests a possible direction.

Before we prove Theorem 1, let's recap an important element about the NDB API in NDB Cluster. The NDB API accepts four different kinds of operations: primary key access, unique key access, full table scan and range scan using an ordered index. So in the case a SQL API node is used, all SQL statements are translated into the four kinds of operations we've just described. The following proof will be based on these operations so that the Theorem applies whether you use the SQL API or the NDB API directly. This effort of finding a method that applies for both the SQL API and the NDB API will prevail throughout the rest of the section when it applies.

        Event of starting the commit phase of T1 →

        Event of releasing the write lock on the primary replica for the write in T1(since it happens in the  commit phase) →

        Event of the item read in T2 →

        Event of starting the commit phase of T2(since starting of the commit phase 'happened after' the  end of the prepare phase and the end of the prepare phase 'happened after' any operation in T2, including the item read).                                                                                          
	  

        Event of starting the commit phase of T1 →

        Event of releasing the write lock on the primary replica for the write in T1 →

        Event of acquiring the write lock on the primary replica for the write in T2 →

        Event of starting the commit phase of T2.
	  

        Event of starting the commit phase of T1 →

        Event of releasing the write lock on the primary replica of the write in T1 →

        Event of the read of a data object of the written tuple to evaluate the predicate in T2 →

        Event of starting the commit phase of T2.
	  

        Event of starting the commit phase of T1 → Event of starting the commit phase of T2.
	  

        Event of starting the commit phase of T→Event of starting the commit phase of T.
	  

	                                                                                                       ## 
	  

This proof hints that in a regular history H, if we could convert a RW conflict into a situation similar to those of Theorem 1's, we would be able to achieve serializability.

For an item RW conflict between T1 and T2, all we have to do is to place a shared read lock to the item read if one hasn't been placed yet. Then an argument similar to the WW conflict case will again lead to:

      Event of starting the commit phase of T1 → Event of starting the commit phase of T2.
	

In the NDB API, this implies changing the lock mode form LM_CommittedRead to LM_Read; in the SQL API, changing a regular select to 'select ... lock in share mode' is sufficient.

The predicate RW conflict case is, however, more complex. In general, an update of the 'decision set' could give rise to either an IN operation of one predicate, or an OUT operation of another predicate; an insertion, however, only leads to a possible IN operation.

Let's first look at the OUT operation case. It turns out OUT operations can also lead to predicate-based conflicts, although common literature usually uses insertions, which is an IN operation, for demonstration. The following example illustrates it.

                            T1                                                                     T2

        start transaction;

        select COUNT(*) from t where id1 > 0 and id1 < 2;

                                                                                         start transaction;

                                                                                         delete from t where id1 = 1;

                                                                                         insert into t values(4,4,4,4);

                                                                                         commit;

        select COUNT(*) from t where id1 > 2 and id1 <6 ;

        commit;

	  

	                                                                                                       ## 
	  

Handling OUT operations is relatively easy. Specifically, if operations following the predicate read are reads in the Read-Committed isolation level, just change the lock mode from LM_CommittedRead to LM_Read in the NDB API or change it to 'select ... lock in share mode' in the SQL API; if on the other hand, these operations are writes, we need to do nothing since the write locks will suffice.

We assume the conflict is from T1 to T2 as usual. According to the type B of the Serializability Theorem, the write w in T2 that changes the match of the predicate is not necessarily the one that updates the version v used in the predicate read. Instead, a few writes that generate versions belonging to the same neighborhood of match change as v does may precede w. Let's assume the sequence of transactions that generate these writes to be Ti through Tj. Since v is inside the match set of the predicate, we may just use the same argument as in the item RW conflict case to guarantee:

      Event of starting the commit phase of Ti → … 
    
      → Event of starting the commit phase of Tj
    
      → Event of starting the commit phase of T2.
	

Here the WW conflicts are between write events that generate the 'decision set' versions.

Eventually we have:

      Event of starting the commit phase of T1 → Event of starting the commit phase of T2.
	

Remark 1: In this approach, we use locks on the reads/writes following a predicate read to push away an OUT operation so that relevant transactions are more isolated. As explained in Remark 1 after Definition 4', the fact that another transaction can't inject such a write operation is the key for the argument to be correct.

Remark 2: At first glance, the method we've just developed may not seem to apply to predicate read in a statement like 'select COUNT(*) in t where …' as in the previous example since there are no reads/writes after the predicate read. It turns out if we replace the first statement in T1 to be 'select COUNT(*) from t where id1 > 0 and id1 < 2 lock in share mode;' in Example 13, T2 will hang at the deletion. That is because this select statement place long read lock for the matching tuples even if the only info accessed is the cardinality of the matching set. So the method applies to this case too.

Remark 3: There is a chance that after the placement of the lock, the predicate RW conflict vanishes. For example, after the placement of the lock, the two transactions are separated far enough so that another statement in T1 perform an OUT operation before T2 got a chance. We'll see an example of this kind in the TPCC benchmark later. The point is that after the lock is in place, whether the predicate RW conflict continue to exist or not, the issue is taken care of.

For an IN operation, things become hefty because the pre-image of the update or insertion is not in the match set. MySQL InnoDB and PostgreSQL both handle this with the so called 'gap lock', which is a special case of so-called 'granular locking'. An exposure of 'granular locking' can be found in the classic book [Gr 93].

However the concept 'gap' becomes a global one in NDB Cluster, a distributed system. When a tuple is inserted into a table, it is hashed to a certain data node so that when tuples are accessed in the same locality, they are processed by different data nodes. This parallelism strategy has been a major contribution to the awesome benchmark profile that NDB Cluster demonstrates. But it also implies if we are looking at two neighboring tuples in the same data node with primary keys '4' and '7' respectively, we don't know what exactly the 'gap' is since there may be a tuple with primary key '6' in some other node. To know exactly what it is, in principle we need to query every node to find out each 'gap' there and consolidate all of them into one. And when this process is happening, it'd be better the 'gap' in each node is not altered. This sounds like it requires global synchronization and it is very expansive. The following theorem suggests we might be able to get around it.

        T1 performing the predicate read →

        Release of l1 in T1 (Condition 1) →

        Acquisition of l2 in T2 →

        T2 releasing write lock of the tuple in the primary replica(Condition 1) →

        Predicate read of the updated/inserted tuple in T1(predicate WR conflict).
	  

        Acquisition of write lock for the updated/inserted tuple in T2 → 

        Release of l2 in T2(Condition 1) →

        Acquisition of l1 in T1 →

        T1 performing the predicate read(Condition 1).
	  

	                                                                                                       ## 
	  

Remark: The placement of the conflicting lock pair might get rid of the predicate RW conflict as in the OUT operation case too. In that case, the issue is resolved automatically and the potential predicate-based conflict is replaced with an item-based one on the 'decision set'.

                     T1                                                           T2

                                                                           start transaction;

                                                                           insert into t values (2,2,2,2);

        start transaction;

        select * from t where id1 > 0 and id1 < 4 lock in share mode;
	  

	                                                                                                       ## 
	  

The 'only if' part of Theorem 2 suggests that for each IN operation induced predicate RW conflict, as long as we place conflicting locks l1 and l2 in the specified places, we have l1 'happened before' l2. This further implies the following relation in the 'happened before' partial order by an argument as in Theorem 1:

      Event of starting the commit phase of T1 → Event of starting the commit phase of T2.                                                               
	

For a Read-Only transaction, the event of starting the commit phase of it may be chosen to be one that 'happened after' the execution of SQL statements(hence 'happened after' any lock acquiring event) and 'happened before' any lock releasing event since two-phase locking is employed by NDB Cluster.

So given a history H in NDB Cluster, for both item and predicate RW conflict between T1 and T2 in H, we've devised a way so that the following condition is satisfied:

For WR and WW conflicts in H, Condition * is also satisfied as in Theorem 1. This implies if there was a conflict loop in H, it would lead to the conclusion that 'happened before' was not strict. This is impossible and H is serializable by the Serializability Theorem. So we have proved the following:

One optimization goal of this article is to minimize locking time. The position of l2 of course is optimal, but how about l1? The following example gives us an affirmative answer.

	                       T1                                                            T2

        start transaction;                                                    

        select * from t where id1 = 2 lock in share mode;

                                                                                start transaction;

                                                                                insert into t values(2,2,2,2);

                                                                                select * from t where id1 = 3 for update;

                                                                                commit;

        select * from t where id1 = 3 lock in share mode;

        commit;
	  

	                                                                                                       ## 
	  

So the location of l1 can't be moved downward, but can it be moved upward? The answer is yes, for both l1 and l2. Actually the following Theorem 2' can be proved.

	                                                                                                       ## 
	  

2.4 Applying the method to other Read-Committed implementations

In this development of a systematic method of applying type B of the Serializability Theorem to NDB Cluster, the choice of time point of commit is crucial. For the arguments to go through, it must be chosen to be after all SQL or NDB API statements are executed, but before any long lock is released. If a Read-Committed implementation also uses a variant of two phase locking, but without acquiring locks for every read like NDB Cluster, the starting point of the second phase naturally serves this purpose. This implies we may also apply the method to MySQL InnoDB's Read-Committed and Repeatable-Read isolation levels to achieve serializability.

For a lock-free system like PostgreSQL's Snapshot Isolation, it doesn't have locks and the previous argument can't be applied. In this case, we associate each transaction T with the timestamp of commit t(let's say even a Read-Only transaction commits, as contrary to some Snapshot Isolation implementations). Then a WR or WW conflict be tween T1 and T2 implies a t1 < t2 since they can't be concurrent. For a RW conflict, materialization or promotion can be used to push two concurrent transactions away so that t1 < t2 remains true. This way a conflict loop cannot form since it would otherwise imply a loop of timestamps of commit, which contradicts with strictness of the temporal order. Notice for this argument to go through, the Monotonicity Condition must hold, namely, clocks can't tick backward.

2.5 An example: the TPCC Benchmark

Next we will apply Theorem 3 to the application in the TPCC benchmark to achieve serializability. TPCC is examined in detail in [Fe 05], where the transactions and conflicts between them are represented as DSG. Although we call DSG a graph, it is really a multi-graph since there may be more than one conflict between any two transactions. [Fe 05] identifies and eliminates the so called 'dangerous structure', two consecutive RW conflicts in DSG, to achieve serializability by aborting a participating conflicting transaction.

In the Read-Committed isolation implemented by NDB Cluster, the story is very different. We in general need to place prevention measures for all RW conflicts in DSG one by one. That is because in a pessimistic system like NDB Cluster, we use locks instead of abortion of transactions to prevent conflicts as we've just described. Actually we don't need to impose prevention measures for every one of those conflicts. We will see that in the following

        The New-Order transaction(Read-Write):

        select w_tax from warehouse where w_id=:w_id;

        select d_tax from district where d_id=:d_id and d_w_id=:w_id;

        select d_next_o_id from district where d_id=:d_id and d_w_id=:w_id;

        update district set d_next_o_id=:d_next_o_id+1 where d_id=:d_id and d_w_id=:w_id;

        select c_discount, c_last, c_credit from customer 
           where c_w_id=:w_id and c_d_id=: c_d_id and c_id=:c_id;

        insert into orders values(:o_id, :d_id, :w_id, :c_id, :datetime, NULL, :o_ol_cnt, :o_all_local);

        insert into new_order values(:o_ld, :d_id, :w_id);

        And for each item we're going to order, we execute the following block of statements in a for loop:

        {    
            select i_price, i_name, i_data from item where i_id=:ol_i_id;

            select s_quantity, s_data, s_dist_01, s_dist_02, s_dist_03, s_dist_04,
               s_dist_05, s_dist_06, s_dist_07, s_dist_08, s_dist_09, s_dist_10
               from stock where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            update stock set s_quantity=:s_quantity where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            update stock set s_ytd=s_ytd+:ol_quantity
               where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            update stock set s_order_cnt=s_order_cnt+1
               where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            if (:ol_supply_w_id !=:w_id) {
               update stock set s_remote_cnt=s_remote_cnt+1
               where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;
            }

            insert into order_line values(:o_id, :d_id, :w_id, :ol_number, :ol_i_id,
               :ol_supply_w_id, NULL, :ol_quantity, :ol_amount, :ol_dist_info);
        }

        The Payment transaction(Read-Write):

        select w_street_1, w_street_2, w_city, w_state, w_zip, w_name 
           from warehouse where w_id=:w_id;

        update warehouse set w_ytd=w_ytd+:h_amount where w_id=:w_id;

        select d_street_1, d_street_2, d_city, d_state, d_zip, d_name 
           from district where d_w_id=:w_id and d_id=:d_id;

        update district set d_ytd=d_ytd+:h_amount where d_w_id=:w_id and d_id=:d_id;

        if the customer making the payment is represented by a name{          //60% chances

            select count(c_id) into :namecnt from customer 
               where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id;

            declare c_byname cursor for 
                select c_first, c_middle, c_id, c_street_1, c_street_2, c_city, c_state, 
                   c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
                   from customer where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
                   order by c_first;

            open c_byname;

            if(:namecnt%2) :namecnt++;
            for (n=0;n < :namecnt/2;n++) {
                fetch c_byname
                   into :c_first, :c_middle, :c_id, :c_street_1, :c_street_2, :c_city, :c_state, 
                   :c_zip, :c_phone, :c_credit, :c_credit_lim, :c_discount, :c_balance, :c_since
            }

            close c_byname;
        }
        else if the customer making the payment is represented by an id{      //40% chances

            select c_first, c_middle, c_last, c_street_1, c_street_2, c_city, c_state, 
               c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
               from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;
        }

        update customer set c_balance=c_balance-:h_amount
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        update customer set c_ytd_payment=c_ytd_payment+:h_amount
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        update customer set c_payment_cnt=c_payment_cnt+1
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        if (:c_credit=”BC”){

            select c_data from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

            update customer set c_data=:c_data
               where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;
        }    

        insert into history values(:c_d_id, :c_w_id, :c_id, :d_id, :w_id, :datatime, :h_amount, :h_data);

        The Order-Status transaction(Read-Only):

        if the customer in the order is represented by a name{                         //60% chances

            select count(c_id) into :namecnt from customer 
               where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id;

            declare c_name cursor for 
               select c_balance, c_first, c_middle, c_id from customer
                  where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
                  order by c_first;

            open c_name;

            if (:namecnt%2) :namecnt++;
            for (n=0;n < :namecnt/2;n++) {
                fetch c_name
                   into :c_balance, :c_first, :c_middle, :c_id;
            }

            close c_name;
        } 
        else if the customer in the order is represented by an id{                    //40% chances

            select c_balance, c_first, c_middle, c_last from customer
               where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;
        }
    
        declare c_order cursor for
           select o_id, o_carrier_id, o_entry_d from orders 
              where o_d_id=:c_d_id and o_w_id=:c_w_id and o_c_id=:c_id
              order by o_id desc;

        open c_order;

        fetch c_order 
           into :o_id, :o_carrier_id, :o_entry_d;

        close c_order;

        select ol_i_id, ol_supply_w_id, ol_quantity, ol_amount, ol_delivery_d from order_line
           where ol_d_id=:c_d_id and ol_w_id=:c_w_id and ol_o_id=:o_id;

        The Delivery transaction(Read-Write):
		
        declare c_no cursor for
           select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
              order by no_o_id asc;

        open c_no;
 
        fetch c_no into :no_o_id;

        close c_no;

        if the former cursor returns a non-empty set{

            delete from new_order where no_d_id=:d_id and no_w_id=:w_id and no_o_id=:no_o_id;

            //In TPCC's delivery transaction, the previous two statements are expressed as an updatable 
            cursor. Since NDB Cluster doesn't support updatable cursors, I've changed it to the 
            previous two statements which are semantically equivalent.         

            select o_c_id from orders where o_d_id=:d_id and o_w_id=:w_id and o_id=:no_o_id;

            update orders set o_carrier_id=:o_carrier_id
               where o_d_id=:d_id and o_w_id=:w_id and o_id=:no_o_id;

            update order_line set ol_delivery_d=:datetime
               where ol_d_id=:d_id and ol_w_id=:w_id and ol_o_id=:no_o_id;

            select sum(ol_amount) from order_line 
               where ol_d_id=:d_id and ol_w_id=:w_id and ol_o_id=:no_o_id;

            update customer set c_balance=c_balance+:ol_total
               where c_id=:c_id and c_d_id=:d_id and c_w_id=:w_id;

            update customer set c_delivery_cnt=c_delivery_cnt+1
               where c_id=:c_id and c_d_id=:d_id and c_w_id=:w_id;
        }

        The Stock-Level transaction(Read-Only):

        select d_next_o_id from district where d_w_id=:w_id and d_id=:d_id; 

        select distinct(ol_i_id) from order_line 
           where ol_w_id=:w_id and ol_d_id=:d_id and ol_o_id<:o_id and ol_o_id>=:o_id-20;

        for each ol_i_id obtained in the last statement, assuming it is stored in an array cell :ol_i_id[i] {

            select count(s_i_id) from stock 
               where s_i_id=:ol_i_id[i] and s_w_id=:w_id and s_quantity<:threshold;
        }                                                                                                 
	  

        select o_id, o_carrier_id, o_entry_d from orders 
           where o_d_id=:c_d_id and o_w_id=:c_w_id and o_c_id=:c_id
           order by o_id desc;                                                                                            
	  

        select * from t_o_n 
           where o_d_id=:c_d_id and o_w_id=:c_w_id and o_c_id=:c_id
           lock in share mode;                                                                                                     
	  

        select o_id, o_carrier_id, o_entry_d from orders 
           where o_d_id=:c_d_id and o_w_id=:c_w_id and o_c_id=:c_id
           order by o_id desc;                                                                                                      
	  

        select * from t_o_n where o_d_id=:d_id and o_w_id=:w_id and o_c_id=:c_id for update;.                                                                                           
	  

        insert into orders values(:o_id, :d_id, :w_id, :c_id, :datetime, NULL, :o_ol_cnt, :o_all_local);.                                                                                              
	  

        select d_next_o_id from district where d_w_id=:w_id and d_id=:d_id lock in share mode;.                                                                                             
	  

        select * from t_s_n
           where s_i_id=:ol_i_id[i] and s_w_id=:w_id
           lock in share mode;                                                                                                   
	  

        select count(s_i_id) from stock 
           where s_i_id=:ol_i_id[i] and s_w_id=:w_id and s_quantity < :threshold;.                                                                                                   
	  

        select * from t_s_n where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id for update;.                                                                                                  
	  

        update stock set s_quantity=:s_quantity where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;.                                                                                                  
	  

        select count(s_i_id) from stock 
           where s_i_id=:ol_i_id[i] and s_w_id=:w_id and s_quantity < :threshold
           lock in share mode;                                                                                                   
	  

        select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
           order by no_o_id asc;
	  

        select * from t_d1_n where no_d_id=:d_id and no_w_id=:w_id lock in share mode;                                                                                                
	  

        select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
           order by no_o_id asc;.                                                                                                  
	  

        select * from t_d1_n where no_d_id=:d_id and no_w_id=:w_id for update;.                                                                                                   
	  

        insert into new_order values(:o_ld, :d_id, :w_id);.                                                                                                    
	  

        select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
           order by no_o_id asc;

        delete from new_order where no_d_id=:d_id and no_w_id=:w_id and no_o_id=:no_o_id;                                                                                                    
	  

        select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
           order by no_o_id asc
           lock in share mode;                                                                                                   
	  

        select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
           order by no_o_id asc
           for update;                                                                                                    
	  

        The New-Order transaction(Read-Write):

        select w_tax from warehouse where w_id=:w_id;

        select d_tax from district where d_id=:d_id and d_w_id=:w_id;

        select d_next_o_id from district where d_id=:d_id and d_w_id=:w_id;

        update district set d_next_o_id=:d_next_o_id+1 where d_id=:d_id and d_w_id=:w_id;

        select c_discount, c_last, c_credit from customer 
           where c_w_id=:w_id and c_d_id=: c_d_id and c_id=:c_id;

        insert into orders values(:o_id, :d_id, :w_id, :c_id, :datetime, NULL, :o_ol_cnt, :o_all_local);

        insert into new_order values(:o_ld, :d_id, :w_id);

        And for each item we're going to order, we execute the following block of statements in a for loop:

        {    
            select i_price, i_name, i_data from item where i_id=:ol_i_id;

            //n->n
            select s_quantity, s_data, s_dist_01, s_dist_02, s_dist_03, s_dist_04,
               s_dist_05, s_dist_06, s_dist_07, s_dist_08, s_dist_09, s_dist_10
               from stock where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id
               lock in share mode;

            update stock set s_quantity=:s_quantity where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            update stock set s_ytd=s_ytd+:ol_quantity
               where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            update stock set s_order_cnt=s_order_cnt+1
               where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            if (:ol_supply_w_id !=:w_id) {
               update stock set s_remote_cnt=s_remote_cnt+1
               where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;
            }

            insert into order_line values(:o_id, :d_id, :w_id, :ol_number, :ol_i_id,
               :ol_supply_w_id, NULL, :ol_quantity, :ol_amount, :ol_dist_info);
        }
		
        //s->n
        select * from t_s_n where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id for update;

        //o->n
        select * from t_o_n where o_d_id=:c_d_id and o_w_id=:c_w_id and o_c_id=:c_id for update;

        //d1->n
        select * from t_d1_n where no_d_id=:d_id and no_w_id=:w_id for update;

        The Payment transaction(Read-Write):

        select w_street_1, w_street_2, w_city, w_state, w_zip, w_name 
           from warehouse where w_id=:w_id;

        update warehouse set w_ytd=w_ytd+:h_amount where w_id=:w_id;

        select d_street_1, d_street_2, d_city, d_state, d_zip, d_name 
           from district where d_w_id=:w_id and d_id=:d_id;

        update district set d_ytd=d_ytd+:h_amount where d_w_id=:w_id and d_id=:d_id;

        if the customer making the payment is represented by a name{          //60% chances

            select count(c_id) into :namecnt from customer 
               where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id;

            //p->p, p->d2
            declare c_byname cursor for
               select c_first, c_middle, c_id, c_street_1, c_street_2, c_city, c_state, 
                  c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
                  from customer where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
                  order by c_first
                  lock in share mode;

            open c_byname;

            if(:namecnt%2) :namecnt++;
            for (n=0;n < :namecnt/2;n++) {
                fetch c_byname
                   into :c_first, :c_middle, :c_id, :c_street_1, :c_street_2, :c_city, :c_state, 
                   :c_zip, :c_phone, :c_credit, :c_credit_lim, :c_discount, :c_balance, :c_since
            }

            close c_byname;
        }
        else if the customer making the payment is represented by an id{      //40% chances
            
            //p->p, p->d2
            select c_first, c_middle, c_last, c_street_1, c_street_2, c_city, c_state, 
               c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
               from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id
               lock in share mode;
        }

        update customer set c_balance=c_balance-:h_amount
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        update customer set c_ytd_payment=c_ytd_payment+:h_amount
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        update customer set c_payment_cnt=c_payment_cnt+1
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        if (:c_credit=”BC”){

            //p->p
            select c_data from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id
               lock in share mode;

            update customer set c_data=:c_data
               where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;
        }    

        insert into history values(:c_d_id, :c_w_id, :c_id, :d_id, :w_id, :datatime, :h_amount, :h_data);

        The Order-Status transaction(Read-Only):

        if the customer in the order is represented by a name{                         //60% chances

            select count(c_id) into :namecnt from customer 
               where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id;

            //o->p and o->d2
            declare c_name cursor for 
               select c_balance, c_first, c_middle, c_id from customer
                  where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
                  order by c_first
                  lock in share mode; 

            open c_name;

            if (:namecnt%2) :namecnt++;
            for (n=0;n < :namecnt/2;n++) {
                fetch c_name
                   into :c_balance, :c_first, :c_middle, :c_id;
            }

            close c_name;
        } 
        else if the customer in the order is represented by an id{                    //40% chances

            //o->p and o->d2
            select c_balance, c_first, c_middle, c_last from customer
               where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id
               lock in share mode;
        }
		
        //o->n
        select * from t_o_n 
           where o_d_id=:c_d_id and o_w_id=:c_w_id and o_c_id=:c_id
           lock in share mode;
		   
        //o->d2
        declare c_order cursor for
           select o_id, o_carrier_id, o_entry_d from orders 
              where o_d_id=:c_d_id and o_w_id=:c_w_id and o_c_id=:c_id
              order by o_id desc
              lock in share mode;

        open c_order;

        fetch c_order 
           into :o_id, :o_carrier_id, :o_entry_d;

        close c_order;
		
        //o->d2
        select ol_i_id, ol_supply_w_id, ol_quantity, ol_amount, ol_delivery_d from order_line
           where ol_d_id=:c_d_id and ol_w_id=:c_w_id and ol_o_id=:o_id
           lock in share mode;

        The Delivery transaction(Read-Write):
		
        //d1->n
        select * from t_d1_n where no_d_id=:d_id and no_w_id=:w_id 
           lock in share mode;
		   
        //The following read lock can be removed if only one delivery transaction with
        //matching NO_W_ID and NO_D_ID is allowed at any given time
        //d2->d2
        declare c_no cursor for
           select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
              order by no_o_id asc
              lock in share mode;

        open c_no;
 
        fetch c_no into :no_o_id;

        close c_no;

        if the former cursor returns a non-empty set{

            delete from new_order where no_d_id=:d_id and no_w_id=:w_id and no_o_id=:no_o_id;

            //In TPCC's delivery transaction, the previous two statements are expressed as an updatable 
            cursor. Since NDB Cluster doesn't support updatable cursors, I've changed it to the 
            previous two statements which are semantically equivalent.         

            select o_c_id from orders where o_d_id=:d_id and o_w_id=:w_id and o_id=:no_o_id;

            update orders set o_carrier_id=:o_carrier_id
               where o_d_id=:d_id and o_w_id=:w_id and o_id=:no_o_id;

            update order_line set ol_delivery_d=:datetime
               where ol_d_id=:d_id and ol_w_id=:w_id and ol_o_id=:no_o_id;

            select sum(ol_amount) from order_line 
               where ol_d_id=:d_id and ol_w_id=:w_id and ol_o_id=:no_o_id;

            update customer set c_balance=c_balance+:ol_total
               where c_id=:c_id and c_d_id=:d_id and c_w_id=:w_id;

            update customer set c_delivery_cnt=c_delivery_cnt+1
               where c_id=:c_id and c_d_id=:d_id and c_w_id=:w_id;
        }

        The Stock-Level transaction(Read-Only):

        //s->n
        select d_next_o_id from district where d_w_id=:w_id and d_id=:d_id 
           lock in share mode;

        select distinct(ol_i_id) from order_line 
           where ol_w_id=:w_id and ol_d_id=:d_id and ol_o_id<:o_id and ol_o_id>=:o_id-20;

        for each ol_i_id obtained in the last statement, assuming it is stored in an array cell :ol_i_id[i] {
		
            //s->n
            select * from t_s_n
               where s_i_id=:ol_i_id[i] and s_w_id=:w_id
               lock in share mode;

            select count(s_i_id) from stock 
               where s_i_id=:ol_i_id[i] and s_w_id=:w_id and s_quantity<:threshold;
        }                                                                                                 
	  

	                                                                                                       ## 
	  

Remark: A helper program to generate the tables in this example is available here

There is one issue though in the way we prevent the predicate RW conflicts in NDB Cluster: we introduce new tables and possible access to them. Does this cause extra conflicts and require more prevention measures? It does introduce new conflicts. For example, inserting and deleting a row in such a table give rise to a WW conflict. It turns out it doesn't require more prevention measures. We will give a proof after we introduce the Ramification Theorem in next section.

2.6 Application of the Generalized Serializability Theorems to TPCC

So this is what we got if we were to apply type B of the Serializability Theorem to the TPCC benchmark. How about the Generalized Serializability Theorems? The Generalized Serializability Theorems suggest that inconsistencies don't show up in subset C and they only show up in its complement, a subset of the Read-Only transactions. Let's name this complement I as in inconsistency. In a real world application, inconsistencies in these Read-Only transactions could be displayed in some sort of monitor and in general are harmless since they won't affect the underlying database. But there are exceptions. Consider the following

	                                                                                                       ## 
	  

So let's now apply the Generalized Serializability Theorems to the TPCC benchmark and see what happens. From the analysis in Example 16, we know that we have prevention measures between the following transaction pairs to avoid a conflict loop to show up: o->p, o->d2, o->n, d1->n, d2->d2, s->n, p->p, n->n. Notice that although d1 is a Read-Only transaction logically, we can't removed the prevention measures for d1->n readily since they are also for d2->n. So if we apply Generalized Serializability Theorem I to it, only four pairs of them remain significant: d1->n, d2->d2, p->p, n->n. In a real world application, if we could arrange the delivery transactions to be in a period of time where new-order transactions are not allowed(for example, for a region, schedule delivery processing at 4 am of that region and suspend incoming orders could be one way to leverage this), we may remove prevention measures for d1->n(and d2->n too) since this way Condition * is guaranteed to be satisfied for this RW conflict. If we further constrain the application at the second-tier to make sure that no two delivery transactions with matching W_ID and D_ID may be active simultaneously, we can get rid of prevention measures for d2->d2 too. This mean we've just proved the following

Remark: In this theorem we've demonstrated not only using Generalized Serializability Theorems, but also rewriting the application, to reduce consistency related overhead while preserving consistency. We call it Theorem 4' instead of Theorem 4 since we still need to take care of the timestamp columns and the statements in the theorem are not exactly correct yet. But it is important to get the idea about what we are trying to achieve here. We'll present Theorem 4 after we cover the Ramification Theorem. So we are getting ahead of ourselves again.

This is not as cool as Fekete's result that TPCC benchmark is serializable under Snapshot Isolation, but it is still awesome since only minimum prevention measures have to be placed in the payment and new-order transactions to achieve desirable consistency. The significance of this result can actually be extended further: with the minimum prevention measures placed, it probably will perform well in a dbt2 like benchmark; and if it does, it suggests that we've not only solved the consistency problem for TPCC under NDB Cluster's Read-Committed isolation level, such a systematic way using Generalized Serializability Theorems will also perform reasonably well for a lot of application under NDB Cluster's Read-Committed isolation level since TPCC itself is a rather complex one. We will try to collect more evidence to support this conclusion down the road.

Somebody might argue that it would not be reasonable to allow possible inconsistencies in the order-status transaction since it would lead to unpleasant customer experience. On the other hand, the situation would be better for the stock-level transaction since a manager may go down to the warehouse to check it if data had shown a dangerous low stock for an important product. If these could be asserted, we need to include back prevention measures for the order-status transaction, namely, those for: o->p, o->d2, o->n. Of course, this implies Generalized Serializability Theorem II is applied.

Remark: This method of using the Generalized Serializability Theorems can certainly be applied to MySQL InnoDB's Read-Committed and Repeatable-Read isolation levels. In Snapshot Isolation like that of PostgreSQL's, a conflict loop is only possible if it contains at least two consecutive RW conflicts. If we could move a Read-Only transaction T from subset C to subset I(or designing your application to be so), we potentially reduced the RW conflicts in subset C since all those started from T were removed. This way maybe we could conclude that the rest of subset C didn't contain a loop and turned out to be serializable while we couldn't have done so with the original subset C. In this approach, we essentially sacrifice the consistency of Read-Only transactions for performance and this could be a good trade-off since the database will remain consistent.

3.1 Issues with timestamp related fields

                  T1                                                        T2

                                                                        start transaction;

        start transaction;

        select a row r1 in t1;

                                                                        update r1 in t1;

                                                                        insert a row r2 into t2 such that a field f
                                                                        of r2 is the value returned from NOW(); 

                                                                        commit;  

        insert a row r3 into t2 such that a field f
        of r3 is the value returned from NOW(); 

        commit;                                                                                          
	  

	                                                                                                       ## 
	  

To understand why this inconsistency would show up, we introduce a type of virtual transaction T and a virtual field named 'system time' into the database. What each transaction T does is to update 'system time' per time resolution of the system and commit. And the NOW() function is implemented as a select to this field. This way, there is an item RW conflict incident on 'system time' from T2 to T, for some T; and there is also an item-based WR conflict on this field from T to T1, for possibly another T(there could be a sequence of Ts in between which forms a sequence of WW conflicts in general). Hence a conflict loop is formed and the inconsistency is explained. Unfortunately, this won't help us much since T is virtual and we can't place prevention measures in it. One possible approach to handle this issue is to take the 'if we can't prevent it, we constrain it' approach and this idea leads to the Ramification Theorem.

Remark: In PostgreSQL's Serializable Snapshot Isolation, the history in Example 18 also executes successfully. The 'start transaction' statement in T2 must precede that of T1's for T2's timestamp to precede T1's since the timestamp represents the time corresponding to the snapshot when the transaction starts. In MySQL InnoDB's serializable isolation level, this history doesn't show up since item read requires a read lock and T2's update would have to wait after T1's commit. But an example in which we insert NOW() into t2 twice from T1 and once from T2 so that T2's insertion is interleaved between T1's duo would also indicate a problem with the same nature. That means the so called serializable isolation levels they've implemented are only correct less the datetime relevant fields. I'll define precisely what this means when I present the Ramification Theorem.

Before we introduce the Ramification Theorem, let's look at a few examples about how inconsistency ramifies from field to field.

	              T1                                                        T2

                                                                    start transaction;

        start transaction;

        select a row r1 in t1;

                                                                    update r1 in t1;

                                                                    insert a row r2 into t2 such that a field f
                                                                    of r2 is the value returned from NOW();

                                                                    select f from r2; 

                                                                    insert a row r4 to t3 such that a field f1 of
                                                                    r4 is the selected value in previous select;                             
                                                                    //The previous two statements can be
                                                                    //combined into a statement like:
                                                                    //insert into t3 values (id, select f from…);

                                                                    commit;  

        insert a row r3 into t2 such that a field f
        of r3 is the value returned from NOW();

        select f from r3; 

        insert a row r5 to t3 such that a field f1 of
        r5 is the selected value in previous select;
        //The previous two statements can be
        //combined into a statement like:
        //insert into t3 values (id, select f from…);

        commit;                                                                    
	  

	                                                                                                       ## 
	  

Inconsistencies not only would ramify to other fields through item reads, but also would do so through predicate reads.

	                T1                                                    T2

                                                                    start transaction;

        start transaction;

        select a row r1 in t1;

                                                                    update r1 in t1;

                                                                    insert a row r2 into t2 such that a field f
                                                                    of r2 is the value returned from NOW();

                                                                    update a field f1 for every tuple in t2
                                                                    where f >= '2019-12-31 00:00:00'
                                                                    and f < '2020-01-01 00:00:00';                           

                                                                    commit;  

        insert a row r3 into t3 such that a field f
        of r3 is the value returned from NOW();

        update a field f1 for every tuple in t3
        where f >= '2019-12-31 00:00:00'
        and f < '2020-01-01 00:00:00';

        commit;                                                                                                
	  

	                                                                                                       ## 
	  

We can see this more clearly in the next example.

	                                                                                                       ## 
	  

Let's look at a more complex case.

	              T1                                                              T2

                                                                            start transaction;
 
        start transaction;

        select a row r1 in t1;
                                                                                         

                                                                            update r1 in t1;

                                                                            insert a row r2 into t2 such that fields f, f1
                                                                            of r2 are the value returned from NOW()
                                                                            and a positive integer respectively;

                                                                            select f1 from t2 where 
                                                                            f >= '2019-12-31 00:00:00'
                                                                            and f < '2020-01-01 00:00:00' 
                                                                            and store it into variable c; 

                                                                            insert a row r4 to t4 such that a field f2 of
                                                                            r4 is c;                             

                                                                            commit;  

        insert a row r3 into t3 such that fields f, f1
        of r3 are the value returned from NOW()
        and a positive integer respectively;

        select f1 from t3 where 
        f >= '2019-12-31 00:00:00'
        and f < '2020-01-01 00:00:00'
        and store it into variable c;

        //assuming c got a default value 0 
        insert a row r5 to t4 such that a field f2 of
        r5 is c;  

        commit;                                                                                              
	  

	                                                                                                       ## 
	  

One interesting thing about this extension is that we can also see the inconsistency from variable c. In other words, if the history was serializable, c in T1 should contain a positive value too, instead of the null value in it. The fact that inconsistency can be temporarily stored in a variable provides a second avenue of ramification through a predicate read. In fact, sometimes the values provided for columns in a predicate is given in a variable. The following extension demonstrates exactly that.

	             T1                                                              T2

                                                                            start transaction;  

        start transaction;

        select a row r1 in t1;

                                                                            update r1 in t1;

                                                                            insert a row r2 into t2 such that fields f, f1
                                                                            of r2 are the value returned from NOW()
                                                                            and a positive integer respectively;

                                                                            select f1 from t2 where 
                                                                            f >= '2019-12-31 00:00:00'
                                                                            and f < '2020-01-01 00:00:00' 
                                                                            and store it into variable c; 

                                                                            update t4 set … where … and col = c;                             

                                                                            commit;  

        insert a row r3 into t3 such that fields f, f1
        of r3 are the value returned from NOW()
        and a positive integer respectively;

        select f1 from t3 where 
        f >= '2019-12-31 00:00:00'
        and f < '2020-01-01 00:00:00'
        and store it into variable c;

        //assuming c got a default value 0
        update t4 set … where … and col = c;

        commit;                                                                                                   
	  

	                                                                                                       ## 
	  

This is how a variable affects the process of identifying which data objects are to access by affecting the predicate read. Since this process could include a cursor sometimes, there is a second way how a variable affects it. The following example in the TPCC application demonstrates this.

        select count(c_id) into :namecnt from customer 
           where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id;

        declare c_byname cursor for 
           select c_first, c_middle, c_id, c_street_1, c_street_2, c_city, c_state, 
              c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
              from customer where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
              order by c_first;

        open cursor c_byname;

        if(namecnt%2) namecnt++;
        for(n=0;namecnt/2;n++)
        {
           fetch c_by_name
              into :c_first, :c_middle, :c_id, :c_street_1, :c_street_2, :c_city, :c_state,
              :c_zip, :c_phone, :c_credit, :c_credit_lim, :c_discount, :c_balance, :c_since; 
        }

        close c_byname;                                                                                              
	  

	                                                                                                       ## 
	  

There is yet another way that inconsistency ramifies, namely, through a branch. The following two extensions demonstrate exactly this.

	              T1                                                                 T2

                                                                            start transaction;

        start transaction;

        select a row r1 in t1;

                                                                            update r1 in t1;

                                                                            insert a row r2 into t2 such that a field f
                                                                            of r2 is the value returned from NOW();

                                                                            select f from r2 and store it into variable c; 
                                                 
                                                                            if (c >= '2019-12-31 00:00:00'
                                                                            and c < '2020-01-01 00:00:00') {

                                                                            insert a row r4 into t4;}                             

                                                                            commit;  

        insert a row r3 into t3 such that a field f
        of r3 is the value returned from NOW();

        select f from r3 and store it into variable c;
 
        if (c >= '2019-12-31 00:00:00'
        and c < '2020-01-01 00:00:00') {

        insert a row r5 into t4;}  

        commit;                                                                                                  
	  

	                                                                                                       ## 
	  

	                T1                                                              T2

                                                                            start transaction;

        start transaction;

        select a row r1 in t1;

                                                                            update r1 in t1;

                                                                            insert a row r2 into t2 such that a field f 
                                                                            of r2 is the value returned from NOW(); 

                                                                            select COUNT(*) from t2 where
                                                                            f >= '2019-12-31 00:00:00'
                                                                            and f < '2020-01-01 00:00:00';

                                                                            //Assuming the result of previous select is
                                                                            //stored in a variable c
                                                                            if c > 0{

                                                                            insert a row r4 into t4;}

                                                                            commit;  

        insert a row r3 into t3 such that a field f
        of r3 is the value returned from NOW();

        select COUNT(*) from t3 where
        f >= '2019-12-31 00:00:00'
        and f < '2020-01-01 00:00:00';

        //Assuming the result of previous select is
        //stored in a variable c
        if c > 0{

        insert a row r5 into t4;}

        commit;                               
	  

	                                                                                                       ## 
	  

3.2 The process of ramification

Based on these extensions and examples we've just discussed, let's try to summarize how inconsistencies can ramify from a base set B of columns and what a ramify operation should be. Firstly, the ramify operation has to read some fields from a base set B and there are obviously two ways of doing that: through an item read or a predicate read. Secondly, for the inconsistencies to ramify to a field outside base set B, a write to that field is a must. What could happen between these two kinds of reads and the write is of our interest.

The first situation is when the read is item-based and the field written is represented as a function F whose variables assumes values returned from those fields read. This is the most common case and a typical scenario is in a statement like 'update col=col+1 …'. Extension #1 serves as a more complex example. In that specific case, f1=F(f) and F is the identity function.

The case that starts with a predicate read is more complex since there are two ways it may come into play. The second situation we are discussing in the next few paragraphs represents one such way and it is demonstrated in Extension #2 and Example 19.

      D := { (Id, x1, x2, …, xn) | Id being the unique identifier of a tuple P evaluates;
      xi, 0 < i <= n, being the values for fields in the 'decision set' of this tuple}.                                                                                               
	

We further define NULL to be a value that doesn't equal to any tuple, and

      I := { tuples uniquely identified by Id} U { NULL}.                                                                                                    
	

Then we may define a map M: D → I as following to represent the predicate read operation:

      M(Id, x1, x2, …, xn) = the tuple identified by Id, if (x1, x2, …, xn) matches P;
      M(Id, x1, x2, …, xn) = NULL, otherwise.                                                                                                    
	

The image Im satisfying M(D) = Im V { NULL }, here V is the disjoint union, is the set of all matching tuples for P. We use this map M to interpret the process of identifying the matching tuples in the predicate read for P.

Under this map, inconsistency inside a set of tuples, which is a characteristics of the tuple set, could become visible in Im. Notice that it is only when the set of fields with inconsistency intersects with the 'decision set' we could possibly see the inconsistency in Im since in that case the inconsistency may or may not be filtered out. It might be better to comprehend this with an analogy: suppose there is a production line which produces a set of 10000 products in a shift; we sample 100 of them to assess the quality of the whole set and a defective product is noticeable only when it falls into that subset of 100 samples.

From Extension #3 we see that inconsistency can be stored in a variable. This leads to the third situation and the second way that a predicate read can be affected by inconsistency. It is demonstrated in Extension #4. The difference between the second way and the first way can be described by the following analogy: view a predicate read as a black box; then to affect its output, you may either affect the input, which is the first way, or affect the black box, which is the second way.

There is more about storing inconsistency in a variable. If it is used in a predicate read of an update or a delete statement, it is the third situation we've just described. If it is used in a deterministic function to write to a field, then it is just the fourth situation and Extension #3 is such an example. Also, if in Example 19, the selected info is stored in a variable before it is written to the field 'total accounts'(say, originally it uses a sub-query so that the selected info is fed into the update statement), it is interpreted as the fourth situation. So now Example 19 can be interpreted in either way: the second situation or the fourth one, depending on the syntax. This applies to Extension #1 too: it may be interpreted as the the fourth or the first situation, depending on whether it uses a variable or not.

The process of identifying which data objects are to access sometimes contains a cursor besides the predicate read. If the variable containing the inconsistency could affect this cursor and the cursor is an updatable one for an update or a delete statement, it is the fifth situation. For example, a variable like :namecnt as in Example 20 that could affect a branching statement for the cursor could be such a case if the cursor turns out to be an updatable one. Also, if the cursor is a scrollable one, the fetch statement of the cursor could be in the following form:

      FETCH ABSOLUTE n FROM cursor_name;

      or

      FETCH RELATIVE n FROM cursor_name;.                                                                                                    
	

Inconsistency in variable n could also affect the process of identifying which data objects are to access and it's considered to be the fifth situation if the cursor happens to be an updatable one.

If on the other hand the variable is used in a branch as demonstrated in Extension #5 and Extension #6, but not in the fifth situation, it is the sixth situation. The only difference between these two extensions is that inconsistency comes from item read in one of them and from predicate read in the other.

Actually there could also be more complex case like after inconsistency is stored in a variable, the variable is used in a predicate read as in the third situation to retrieve a field and this field is in turn stored in another variable, and this second variable is used in a deterministic function to write a field. And before the final field write this process could happen multiple times and more variables can be involved. In other words, inconsistency can be passed on in the form of a variable chain before touching down to a field in the database and each of these passing is just like the third, the fourth, the fifth or the sixth situation except that a new variable, instead of a field, is written. But if we just look at the last variable(assuming the inconsistency is still visible in this last variable), it can always be reduced into one of the last four situations.

Notice that the case in Example 20 is not summarized in the last four situations since the process it affects is not from an update statement and hence it can't be the last variable in the chain. But it can be a variable in the chain in general. Also since NDB Cluster supports neither updatable or scrollable cursor, the fifth situation will never show up for the application to NDB Cluster in this article. But for systems that do, including the fifth situation is necessary.

This classification is summarized in the following

	                                                                                                       ## 
	  

Remark 1: One could actually view the third, the fifth and the sixth situation as one similar to the fourth one as follows. In the third situation, if a variable affects the predicate read, a value in the variable is first used to identify a predicate read(the map M) and the predicate read is used to identify the matching set. The matching set decides then which tuples to update. Notice here the variable doesn't decide the values to be updated(Another variable could do that job). Its role is more like a variable in a piece-wise function that decides what definition the function might assume under different circumstances. In the sixth situation, a branch is basically a map that decides which set of instructions is executed and which set is not. In our specific case, this set of instructions represents a set of database update statements. It will decide which set(like empty or non-empty set as in Extension #5 and Extension #6) of tuples to update, just like the third situation. The fifth situation is the other way to affect the process of identifying which data objects are to access. The map is one that maps the variable to a subset in the matching set of the associated predicate read.

Remark 2: For the way of passing inconsistency from one variable x to the next variable y inside the chain of variables, there is a similar break down: if this passing doesn't involve any SQL statement, there are two cases. Case 1: it is just a regular mapping so that y is the result and x is a variable of this map. Case 2: x is used in the branching condition of a branch and y is modified in that branch. If on the other hand it involves SQL statements, there are four cases. Case 1: x is used in the process of identifying which data objects are to access(in the predicate read or the cursor that follows) of a select and one of the selected field is used in a mapping whose result is stored in y. Case 2: x is used in the process of identifying which data objects are to access(in the predicate read or the cursor that follows) of a select, an update or a delete and a characteristic of the matching set, like the cardinality, is used in a mapping whose result is stored in y. Case 3: x is used in the branching condition of a branch and one of the selected field in a select in this branch is used in a mapping whose result is stored in y. Case 4: x is used in the branching condition of a branch and a select, an update or a delete is executed in this branch; a characteristic of the matching set, like the cardinality, is used in a mapping whose result is stored in y. And of course, all these cases can, again, be interpreted as a map like in Remark 1.

Remark 3: The structure in Lemma 1 – the leading read, the trailing write and the possible sequence in between – is a generic one. It describes how a field(the one read) affects another(the one written) in a transaction and hence in a history in general. Lemma 1 only emphasizes the case when inconsistency is visible in the field read and the field written. One can also see how the field read affects a variable in the chain if one is present in this structure. We'll use this generic structure to define a ramification operator R shortly.

There is one more issue though. So everything can be interpreted as a map of some sort. This is certainly true if the functions involved are deterministic ones. What if there are non-deterministic functions involved in the process in concern? Can inconsistency ramify through them? Well, all the so-called random functions are actually pseudo-random and so they are really deterministic. Hence we are fine with it.

Notice the discussion surrounding Lemma 1 is a syntactic approach and this comes with two implications. First, if in the future the syntax of SQL is changed, we might need to adjust Lemma 1 and Remark 2 so that they remain correct. For example, in the SQL:2003 standard, an updatable cursor is accessed with the following statements after a fetch statement has positioned the cursor on a specific row:

      UPDATE table_name
         SET …
         WHERE CURRENT OF cursor_name;

      or

      DELETE FROM table_name
         WHERE CURRENT OF cursor_name;.                                                                                                     
	

If in the future the syntax was changed to something like

      UPDATE table_name
         SET …
         WHERE ABSOLUTE n FROM cursor_name;                                                                                                  
	

with the original fetch statement being skipped, we need to add this case to the fifth situation since variable n could contain inconsistency then.

Besides this imaginary example, let's take a look at one that is more realistic. MySQL recently starts to support derived table, which is a kind of sub-query in the FROM clause in the outer query. The following statement is one such example.

      SELECT sb1 into :y, sb2, sb3
         FROM (SELECT s1 AS sb1, s2 AS sb2, s3=:x AS sb3 FROM t1) AS sb
         WHERE sb1 > 1;                                                                                                    
	

In this statement, if inconsistency is visible in variable x, it is possible that it would ramify to variable y. Such a statement would not show up in the current framework since we've assumed every statement only involves one table. But if we were to extend this framework to involve more complex statements like joins, sub-queries and unions, the previous statement must be included. Then none of the four cases in Remark 2 involving a SQL statement applies here since x is not in the process of identifying which data objects are to access of the outer query; it is in the process of identifying which data objects are to access of the sub-query, but the result of the sub-query is not stored in y. So we must provide extra cases in Remark 2 to elaborate this. Right now, the outer operation of a derived table must be a query. If in the future MySQL were to support update and delete for the outer operation too, a new situation must be added to Lemma 1 to reflect this too.

Second, some dialects of SQL might introduce non-standard syntax into cursor related statements and Lemma 1 might need to be adjusted to reflect that. For example, the following is a snippet for demonstration of using an updatable cursor in Oracle and it doesn't resemble the syntax in SQL:2003 standard at all.

      DECLARE
        -- customer cursor
        CURSOR c_customers IS 
            SELECT 
                customer_id, 
                name, 
                credit_limit
            FROM 
                customers
            WHERE 
                credit_limit > 0 
            FOR UPDATE OF credit_limit;
        -- local variables
        l_order_count PLS_INTEGER := 0;
        l_increment   PLS_INTEGER := 0;
    
      BEGIN
        FOR r_customer IN c_customers
        LOOP
            -- get the number of orders of the customer
            SELECT COUNT(*)
            INTO l_order_count
            FROM orders
            WHERE customer_id = r_customer.customer_id;
            -- 
            IF l_order_count >= 5 THEN
                l_increment := 5;
            ELSIF l_order_count < 5 AND l_order_count >=2 THEN
                l_increment := 2;
            ELSIF l_increment = 1 THEN
                l_increment := 1;
            ELSE 
                l_increment := 0;
            END IF;
        
            IF l_increment > 0 THEN
                -- update the credit limit
                UPDATE 
                    customers
                SET 
                    credit_limit = credit_limit * ( 1 +  l_increment/ 100)
                WHERE 
                    customer_id = r_customer.customer_id;
            
                -- show the customers whose credits are increased
                dbms_output.put_line('Increase credit for customer ' 
                    || r_customer.NAME || ' by ' 
                    || l_increment || '%' );
            END IF;
        END LOOP;
    
        EXCEPTION
            WHEN OTHERS THEN
                dbms_output.put_line('Error code:' || SQLCODE);
                dbms_output.put_line('Error message:' || sqlerrm);
                RAISE;
            
      END;                                                                                                     
	

As we can see from the definition, B is always a subset of R(B). If we apply R to B repeatedly, we got a sequence: B, R(B), R(R(B)), … such that any set in the sequence is a subset of a latter one. This expansion has to stop at some point since the set of columns in the database is finite. We call this super-set of all members in the sequence Ram(B), or the ramification set of B. Ram(B) represents the set of columns that could be affected by columns from B when the application executes.

	                                                                                                       ## 
	  

	                                                                                                       (To be continued...)                    
	  

Remark: In the description of the transactions in the TPCC specification, c_data doesn't depend on a datetime value. So according to the description, B = {c_since, h_date, o_entry_d, ol_delivery_d} and Ram(B) = B. We stick to the implementation, not the description, in this case so that Ram(B)' is absolutely consistent.

3.3 Split of transactions to Ram(B)'

Next we try to restrict transactions to Ram(B)' and we will use examples to demonstrate what exactly this means.

Let's first consider the following update statement in a transaction:

      update t set col3=..., col4=..., col5=... where col1=... and col2=...;.                                                                                                    
	

Assuming at least one of col3, col4 and col5 is in Ram(B)'. Suppose col3 and col4 belong to Ram(B)', but col5 belongs to Ram(B). By Lemma 2, col1 and col2 must both be in Ram(B)'. We split the statement into the following two:

      update t set col3=..., col4=... where col1=... and col2=...;

      update t set col5=... where col1=... and col2=...;.                                                                                         
	

We call the first statement the split of the original statement to Ram(B)' since all the fields involved are in Ram(B)'. This operation is designated as a type I split operation and the resulting statement is of course designated as a type I split statement. If all of col3, col4 and col5 are in Ram(B)', we don't need any action since it is already split. The case where all of col3, col4 and col5 are in Ram(B) will be handled shortly,

Now let's look at the following query statement:

      select col3, col4, col5 from t where col1=... and col2=...;.                                                                                                     
	

Assuming at least one of col3, col4 and col5 is in Ram(B)'. Notice that unlike the type I case, some of the selected fields being in Ram(B)' don't necessarily lead col1 and col2 to be so. But by lemma 2: if col1 or col2 is in Ram(B), the selected fields won't affect writes in Ram(B)' at all even if they are in Ram(B)' since otherwise Ram(B) could be further ramified. And for the same reason, the selected fields in Ram(B) won't affect writes to Ram(B)' in any case. So if we want to split those reads that can affect writes to Ram(B)', we only need to consider statements with both col1 and col2 being in Ram(B)' and isolate those selected fields in Ram(B)'. Again let's assume that col3 and col4 belong to Ram(B)', but col5 belongs to Ram(B). Then we split it to the following two:

      select col3, col4 from t where col1=... and col2=...;

      select col5 from t where col1=... and col2=...;.                                                                                                     
	

We call the first statement the split of the original statement to Ram(B)'. This operation is designated as a type II split operation and the resulting statement is of course designated as a type II split statement. That is the part of the original statement that could affect writes to Ram(B)'. Again if all of col3, col4 and col5 are in Ram(B)', it is already split. The case where all of col3, col4 and col5 are in Ram(B) will be handled shortly,

Remark: Speaking of splitting select statements, in an update statement like 'update t set col=col+1 where …', there is a hidden select statement 'select col from t where...' inside. We certainly could split it and classify it as a new type of split statement. But it is OK just to leave it in the type I context since the split of an update statement has done the job.

For a deletion, its 'decision set' can't contain any field in Ram(B) by Lemma 2 unless all the deleted fields are in Ram(B). The later case certainly can't affect writes in Ram(B)' and we designate the former case as a type III split operation and its split consists only of deleted fields in Ram(B)'. The resulting statement is of course designated as a type III split statement.

For an insertion, only the inserted fields in Ram(B)' are considered to be writes that are split from the original statement. This operation is designated as a type IV split operation and the resulting statement is of course designated as a type IV split statement. In general, for statements without a predicate read(without a WHERE clause), the split consists of all the accessed fields in Ram(B)' for all four types since Ram(B) doesn't have a chance to affect a predicate read.

For statements that only consist of a predicate read like 'select COUNT(*) from …', similar to the type II case, we should only consider those statements whose 'decision set' lies completely in Ram(B)'. Although this type of statement is already split, we call it a type V split operation and the resulting statement a type V split statement anyway.

Now let's revisit the following query statement as in type II split operation case:

      select col3, col4, col5 from t where col1=... and col2=...;.                                                                                        
	

This time we assume that all of col3, col4 and col5 are in Ram(B). In this case, if both col1 and col2 are in Ram(B)', the predicate read can still affect writes in Ram(B)', like with its cardinality as we've described after Remark 2 of Lemma 1. Since the current syntax of SQL doesn't allow us to isolate the predicate read from the item reads that follow, we have to include the whole statement as the split statement, but only its predicate read matters of course.

Similar arguments apply to type I and type III operations too. We designate all three as a type VI split operation and the resulting statement is of course designated as a type VI split statement. This is the only type of split statement that includes fields in Ram(B). Notice that in the type I, type II or type III split operation case, if it is the predicate read that could affect writes in Ram(B)', it is already included in the split statement and nothing else needs to be done.

Remark: We've just discussed how to split six relatively simple types of statements here. This means to apply the Ramification Theorems we'll present, we have to re-write our transactions to consist of only these six simple types of statements. Fortunately most, if not all, SQL statements can be re-written into these six simple types.

We call these six types of operation the split operations and the resulting statements the split statements. Also, the split of a set of SQL statements(for example, the set inside a transaction) is just simply the collection of the split of each individual statement in that set.

In a transaction, these split statements, cursors in the process of identifying which data objects are to access, other branches and other non-SQL statements collectively form the split of a transaction to Ram(B)'. Given a history H, we called the collection of splits of member transactions of H on Ram(B)' to be the split of H to Ram(B)'. For the split of H, we may only consider the conflicts between split statements since we have the following observations:

1. All field writes in Ram(B)' are written by type I, type III and type IV split statements.
2. Fields read in type I and type II split statements are the only kind of item reads that can affect field writes in Ram(B)', although not every one of them does. When they do, they might be through the first, the third, the fourth, the fifth or the sixth situation discussed above(To be specific, type I is through the first situation, while type II is through the first, the third, the fourth, the fifth or the sixth situation). Predicate reads in type I, type II, type III, type V and type VI split statements are the only kind of predicate reads that can affect field writes in Ram(B)', although not every one of them does. When they do, they might be through the second, the third, the fourth, the fifth or the sixth situation discussed above(To be specific, type I and type III are through the second situation, while type II, type V and type VI are through the third, the fourth, the fifth or the sixth situation).
3. For any branch that can affect field writes to Ram(B)', its branching condition cannot be affected by Ram(B). In other words, its branching condition can't contain a variable which might store inconsistency originating from Ram(B). Similarly, for a variable in a cursor in the process of identifying which data objects are to access that can affect writes to Ram(B)', it can't be affected by Ram(B) either. Consequently, it can only be affected by Ram(B)'
4. For reads in the split to Ram(B)' that can't affect writes to Ram(B)', they could be affected by Ram(B) in general. There is also a break down about how Ram(B) can affect reads in the split to Ram(B)' like the six situations in Lemma 1: it starts with a read in Ram(B) and inconsistency is stored in a variable, then it is passed down a chain of variables as usual and eventually affects a read in Ram(B)'. If we look at the last variable in the chain, the first type is through the process of identifying which data objects are to access(the predicate read or the cursor that follows), just like the third or the fifth situation except it doesn't end with a write. The second type is when the last variable affects a branch that is enclosing the target read. This is like the sixth situation except it doesn't end with a write. These reads in the split to Ram(B)' that can't affect writes to Ram(B)', but can be affected by Ram(B) are of type II, type V or type VI split statements.

Observations 1 and 2 just follow from the definition of the split operations and Lemmas 1 & 2. Observations 3 does require some explanation.

In Remark 3 of Lemma 1 we know that the starting read, trailing write and the possible chain of variables in between is structural. Hence a branch that can affect field writes to Ram(B)' just means that it is a branch in such a structure, which touches down with a write to a field in Ram(B)'. If this structure also started with a read from Ram(B), it would form the path described in Lemma 1 and Ram(B) would affect Ram(B)', which contradicts with Lemma 2. A similar argument is true for a variable in a cursor in the process of identifying which data objects are to access that could affect writes to Ram(B)'. Hence observation 3 follows.

Observation 4 should be self-explanatory. Notice that a select could have selected fields in Ram(B)' while the predicate read contains fields in Ram(B). This kind of select is not included in observation 4 since an item read in Ram(B)' is not the same as an item read in the split to Ram(B)'.

3.4 The Ramification Theorems

With these four observations, if no split statements as in observation 4 exist, the split on Ram(B)' executes as if Ram(B) didn't exist; if on the other hand, split statements as in observation 4 actually exist, the updates to Ram(B)' are still completely decided by the split. Now we are ready to present the following

	                                                                                                       ## 
	  

We may actually prove that any field written on Ram(B)' can be expressed as a function where the variables contain only fields from Ram(B)' and constants in both cases. When we say 'C doesn't contain a conflict loop', we mean that there is no conflict loop in the set of associated transactions of C for conflicts caused by reads and writes in C. And 'a serial execution of itself' can be achieved by executing the set of associated transactions serially.

Remark 1: When B is empty, case 2 will not show up and the Ramification Theorem becomes the Serializability Theorem. So the Ramification Theorem is a generalization of the Serializability Theorem.

Then the fact that the proof is similar to that of the Serializability Theorems is not a surprise. The key is to isolate the statements that could affect the database portion represented by Ram(B)', which translates into the six types of split statements, and the rest is taken care of by the definition of Ram(B)'. Notice that if we were still interpreting conflicts at the tuple level, it might hardly be natural to come across the definition of split operations and probably wouldn't be able to discover the Ramification Theorem. This is one more piece of evidence that trying to take everything down to the field level would be beneficial. In particular, we don't have a Ramification Theorem in type A of the Serializability Theorem setting unless we introduce concepts like a sub-tuple that covers fields in a table in Ram(B)' and such. We are not going to do that since again, type A is for demonstration only.

Remark 2: From the proof of the Ramification Theorem we know that the split statements contain all item writes to Ram(B)'. They don't contain all item reads to Ram(B)' though. The field reads in a query with 'decision set' of the predicate read intersecting Ram(B) may all be in Ram(B)', but the query itself can't give rise to any split operations. These item reads may contain inconsistencies in general.

	                                                                                                       ## 
	  

	                                                                                                       (To be continued...)                    
	  

The Ramification Theorem is based on type B/C of the Serializability Theorem. The next few theorems combine the Ramification Theorem with type B/C of the Generalized Serializability Theorems.

	                                                                                                       ## 
	  

Remark: Ramification Theorem I is a generalization of Generalized Serializability Theorem I.

	                                                                                                       ## 
	  

Now we are ready to remedy Theorem 4' and present the following

	                                                                                                       ## 
	  

	                                                                                                       ## 
	  

Remark: Ramification Theorem II is a generalization of Generalized Serializability Theorem II.

Ramification Theorem II is designed for the case when some of the Read-Only transactions are important that inconsistencies should not show up in their reads either.

	                                                                                                       ## 
	  

The key observation about the Generalized Serializability Theorems and the Ramification Theorems is that after we rule out those reads that don't affect writes to the database(or a potion of it), we can still prove the rest to be serializable if it is free of conflict loops and hence consistency in the underlying database is preserved. To push this idea to its extreme, consider a subset R of C as in Generalized Serializability Theorem II which consists of reads that don't affect writes to the database. Notice that any read in a Read-Only transaction qualifies as a member of R here. So although in Generalized Serializability Theorem II we require a Read-Only transaction to be consistent or not as a whole, in the following Generalized Serializability Theorem III, we may choose only part of it to be so. But the intended statements in R are usually those in the Read-Write transactions.

	                                                                                                       ## 
	  

	                                                                                                       ## 
	  

	                                                                                                       ## 
	  

Remark: Ramification Theorem III is a generalization of Generalized Serializability Theorem III.

	                                                                                                       ## 
	  

In Ramification Theorem III, three types of reads that do not affect writes to Ram(B)' could be excluded. The first type of reads are those from the Read-Only transactions. They don't affect writes at all, so in particular they don't affect writes to Ram(B)'. The second type of reads are those in Read-Write transactions that don't affect writes and in particular they don't affect writes to Ram(B)'. These two types could both show up in R. The third type of reads are those from observation 4. Although three types of reads have been filtered out, there could still be reads that would not affect writes to Ram(B)' in the remaining statements. Consider reads in the split of C minus R to Ram(B)' that can't affect writes to Ram(B)' and can't be affected by Ram(B) either, but can affect writes in Ram(B). These reads can't be any of the three types excluded already. So in principle we may develop a Ramification Theorem IV that excludes them too and try to push everything beyond the extreme. But the thing is that we develop the Ramification Theorems to handle the issue with datetime fields and it is hard to imagine non-datetime field reads can affect a datetime field write. Such reads can, however, affect writes to the ramification set of the datetime fields. But for the purpose of this article, we might want to settle with Ramification Theorem III for now.

        The New-Order transaction(Read-Write):

        select w_tax from warehouse where w_id=:w_id;

        select d_tax from district where d_id=:d_id and d_w_id=:w_id;

        select d_next_o_id from district where d_id=:d_id and d_w_id=:w_id;
		
        :o_id=:d_next_o_id;

        update district set d_next_o_id=:d_next_o_id+1 where d_id=:d_id and d_w_id=:w_id;

        select c_discount, c_last, c_credit from customer 
           where c_w_id=:w_id and c_d_id=: c_d_id and c_id=:c_id;

        insert into orders values(:o_id, :d_id, :w_id, :c_id, :datetime, NULL, :o_ol_cnt, :o_all_local);

        insert into new_order values(:o_ld, :d_id, :w_id);

        And for each item we're going to order, we execute the following block of statements in a for loop:

        {    
            select i_price, i_name, i_data from item where i_id=:ol_i_id;

            //n->n
            select s_quantity, s_data, s_dist_01, s_dist_02, s_dist_03, s_dist_04,
               s_dist_05, s_dist_06, s_dist_07, s_dist_08, s_dist_09, s_dist_10
               from stock where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id
               lock in share mode;

            update stock set s_quantity=:s_quantity where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            update stock set s_ytd=s_ytd+:ol_quantity
               where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            update stock set s_order_cnt=s_order_cnt+1
               where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;

            if (:ol_supply_w_id !=:w_id) {
               update stock set s_remote_cnt=s_remote_cnt+1
               where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;
            }
			
            :ol_amount = :ol_quantity * :i_price * (1+:w_tax+:d_tax) * (1-:c_discount);

            insert into order_line values(:o_id, :d_id, :w_id, :ol_number, :ol_i_id,
               :ol_supply_w_id, NULL, :ol_quantity, :ol_amount, :ol_dist_info);
        }
		
        //s->n
        select * from t_s_n where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id for update;

        //o->n
        select * from t_o_n where o_d_id=:c_d_id and o_w_id=:c_w_id and o_c_id=:c_id for update;

        //d1->n
        select * from t_d1_n where no_d_id=:d_id and no_w_id=:w_id for update;
		
	  

        select d_next_o_id from district where d_id=:d_id and d_w_id=:w_id;.
	  

        update district set d_next_o_id=:d_next_o_id+1 where d_id=:d_id and d_w_id=:w_id;.
	  

        select s_quantity, s_data, s_dist_01,  s_dist_02,  s_dist_03,  s_dist_04,
           s_dist_05,  s_dist_06,  s_dist_07,  s_dist_08,  s_dist_09,  s_dist_10
           from stock where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id
           lock in share mode;.
	  

        select i_price, i_name, i_data from item where i_id=:ol_i_id;.
	  

        select w_tax from warehouse where w_id=:w_id;.
	  

        select d_tax from district where d_id=:d_id and d_w_id=:w_id;.
	  

        select c_discount, c_last, c_credit from customer 
           where c_w_id=:w_id and c_d_id=: c_d_id and c_id=:c_id;.
	  

        select c_last, c_credit from customer 
           where c_w_id=:w_id and c_d_id=: c_d_id and c_id=:c_id;

        select i_name, i_data from item where i_id=:ol_i_id;

        select s_data, s_dist_01, s_dist_02, s_dist_03, s_dist_04,
           s_dist_05, s_dist_06, s_dist_07, s_dist_08, s_dist_09, s_dist_10
           from stock where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id;.
	  

        update district set d_next_o_id=d_next_o_id+1 where d_id=:d_id and d_w_id=:w_id;.
	  

	                                                                                                       (To be continued...)                    
	  

	            T1                                                               T2

        start transaction;

        select f in tu;

                                                                            start transaction;

                                                                            update f in tu;

                                                                            commit;

        select f in tu again;

        update f in tu;

        commit;                                                                                                               
	  

	                                                                                                       ##                   
	  

Remark: This is an example which demonstrates that we can improve performance by using the Generalized Serializability Theorems or the Ramification Theorems in place of the Serializability Theorem if consistency in some reads can be sacrificed too. We may also use predicate reads to construct a similar example. In general, NOT viewing two identical reads, both item and predicate ones, in a transaction as the same may give us advantages in performance.

        The Payment transaction(Read-Write):

        select w_street_1, w_street_2, w_city, w_state, w_zip, w_name 
           from warehouse where w_id=:w_id;

        update warehouse set w_ytd=w_ytd+:h_amount where w_id=:w_id;

        select d_street_1, d_street_2, d_city, d_state, d_zip, d_name 
           from district where d_w_id=:w_id and d_id=:d_id;

        update district set d_ytd=d_ytd+:h_amount where d_w_id=:w_id and d_id=:d_id;

        if the customer making the payment is represented by a name{          //60% chances

            select count(c_id) into :namecnt from customer 
               where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id;

            //p->p, p->d2
            declare c_byname cursor for
               select c_first, c_middle, c_id, c_street_1, c_street_2, c_city, c_state, 
                  c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
                  from customer where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
                  order by c_first
                  lock in share mode;

            open c_byname;

            if(:namecnt%2) :namecnt++;
            for (n=0;n < :namecnt/2;n++) {
                fetch c_byname
                   into :c_first, :c_middle, :c_id, :c_street_1, :c_street_2, :c_city, :c_state, 
                   :c_zip, :c_phone, :c_credit, :c_credit_lim, :c_discount, :c_balance, :c_since
            }

            close c_byname;
        }
        else if the customer making the payment is represented by an id{      //40% chances
            
            //p->p, p->d2
            select c_first, c_middle, c_last, c_street_1, c_street_2, c_city, c_state, 
               c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
               from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id
               lock in share mode;
        }

        update customer set c_balance=c_balance-:h_amount
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        update customer set c_ytd_payment=c_ytd_payment+:h_amount
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        update customer set c_payment_cnt=c_payment_cnt+1
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        if (:c_credit=”BC”){

            //p->p
            select c_data from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id
               lock in share mode;
			   
            //the + is string concatenation operator, n is the length of the string concatenated to :c_data.
            //Here the implementation instead of the description of the transaction is followed, :h_date            
            //and :h_data are also concatenated to :c_data. The prevention measure for p->p case are
            //also removed for this reason.
            :c_new_data=':c_id, :c_d_id, :c_w_id, :d_id, :w_id, :h_amount, :h_date, :h_data' + :c_data
            :c_data=right-shift(:c_new_data, n), n being the length of the prepended string

            update customer set c_data=:c_data
               where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;
        }

        strncpy(:h_data, :w_name, 10);
        :h_data[10]=' ';
        :h_data[11]=' ';
        :h_data[12]=' ';
        :h_data[13]=' '; 
        strncat(:h_data, :d_name, 10);
        :h_data[24]='\0';		

        insert into history values(:c_d_id, :c_w_id, :c_id, :d_id, :w_id, :datatime, :h_amount, :h_data);
	  

        update warehouse set w_ytd=w_ytd+:h_amount where w_id=:w_id;.                                                                                                                
	  

        update district set d_ytd=d_ytd+:h_amount where d_w_id=:w_id and d_id=:d_id;.                                                                                                                
	  

        select count(c_id) into :namecnt from customer 
           where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id;.                                                                                                                  
	  

        select c_first, c_middle, c_id, c_street_1, c_street_2, c_city, c_state, 
           c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
           from customer where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
           order by c_first;.                                                                                                                 
	  

        select c_first, c_middle, c_last, c_street_1, c_street_2, c_city, c_state, 
           c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
           from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;.                                                                                                                  
	  

        select c_data from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id
           lock in share mode;                                                                                                                
	  

        update customer set c_data=:c_data
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;.                                                                                                                  
	  

        select w_street_1, w_street_2, w_city, w_state, w_zip, w_name 
           from warehouse where w_id=:w_id;

        select d_street_1, d_street_2, d_city, d_state, d_zip, d_name 
           from district where d_w_id=:w_id and d_id=:d_id;.                                                                                                                 
	  

        select w_street_1, w_street_2, w_city, w_state, w_zip 
           from warehouse where w_id=:w_id;

        select d_street_1, d_street_2, d_city, d_state, d_zip 
           from district where d_w_id=:w_id and d_id=:d_id;

        select c_first, c_middle, c_street_1, c_street_2, c_city, c_state, 
           c_zip, c_phone, c_credit_lim, c_discount, c_since
           from customer where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
           order by c_first;

        select c_first, c_middle, c_last, c_street_1, c_street_2, c_city, c_state, 
           c_zip, c_phone, c_credit_lim, c_discount, c_since
           from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;.                                                                                                                
	  

        select c_data from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;

        update customer set c_data=:c_data
           where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id;.                                                                                                                
	  

        The Delivery transaction(Read-Write):
		
        //d1->n
        select * from t_d1_n where no_d_id=:d_id and no_w_id=:w_id 
           lock in share mode;
		   
        //The following read lock can be removed if only one delivery transaction with
        //matching NO_W_ID and NO_D_ID is allowed at any given time
        //d2->d2
        declare c_no cursor for
           select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
              order by no_o_id asc
              lock in share mode;

        open c_no;
 
        fetch c_no into :no_o_id;

        close c_no;

        if the former cursor returns a non-empty set{

            delete from new_order where no_d_id=:d_id and no_w_id=:w_id and no_o_id=:no_o_id;

            //In TPCC's delivery transaction, the previous two statements are expressed as an updatable 
            cursor. Since NDB Cluster doesn't support updatable cursors, I've changed it to the 
            previous two statements which are semantically equivalent.         

            select o_c_id from orders where o_d_id=:d_id and o_w_id=:w_id and o_id=:no_o_id;

            update orders set o_carrier_id=:o_carrier_id
               where o_d_id=:d_id and o_w_id=:w_id and o_id=:no_o_id;

            update order_line set ol_delivery_d=:datetime
               where ol_d_id=:d_id and ol_w_id=:w_id and ol_o_id=:no_o_id;

            select sum(ol_amount) from order_line 
               where ol_d_id=:d_id and ol_w_id=:w_id and ol_o_id=:no_o_id;

            update customer set c_balance=c_balance+:ol_total
               where c_id=:c_id and c_d_id=:d_id and c_w_id=:w_id;

            update customer set c_delivery_cnt=c_delivery_cnt+1
               where c_id=:c_id and c_d_id=:d_id and c_w_id=:w_id;
        }                                                                                                                     
	  

        select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
           order by no_o_id asc;                                                                                                                  
	  

        select o_c_id from orders where o_d_id=:d_id and o_w_id=:w_id and o_id=:no_o_id;.                                                                                                                  
	  

        select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
           order by no_o_id asc;.                                                                                                                   
	  

        select sum(ol_amount) into :ol_total from order_line 
           where ol_d_id=:d_id and ol_w_id=:w_id and ol_o_id=:no_o_id;.                                                                                                                  
	  

        select no_o_id from new_order where no_d_id=:d_id and no_w_id=:w_id
           order by no_o_id asc;.                                                                                                                   
	  

        //p->p, p->d2
        declare c_byname cursor for 
           select c_first, c_middle, c_id, c_street_1, c_street_2, c_city, c_state, 
              c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
              from customer where c_last=:c_last and c_d_id=:c_d_id and c_w_id=:c_w_id
              order by c_first
              lock in share mode;

        //p->p, p->d2
        select c_first, c_middle, c_last, c_street_1, c_street_2, c_city, c_state, 
           c_zip, c_phone, c_credit, c_credit_lim, c_discount, c_balance, c_since
           from customer where c_id=:c_id and c_d_id=:c_d_id and c_w_id=:c_w_id
           lock in share mode;.                                                                                                                   
	  

        //n->n
        select s_quantity, s_data, s_dist_01, s_dist_02, s_dist_03, s_dist_04,
           s_dist_05, s_dist_06, s_dist_07, s_dist_08, s_dist_09, s_dist_10
           from stock where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id
           lock in share mode;.                                                                                                                 
	  

	                                                                                                       (To be continued...)                    
	  

In general, if there is an item read for a field that doesn't affect writes in a transaction and there is a write of that field in another transaction, it is possible that we may use Ramification Theorem III to get rid of the read lock on that field. This is demonstrated in the following

	           T1                                                         T2

        start transaction;

        select f1 from t1;

                                                                    start transaction;

                                                                    update t1 set f1 = …;

                                                                    update t2 set f2 = …;

                                                                    commit;

        update t2 set f2 = …;

        commit;                                                                                                                 
	  

	                                                                                                       ##                   
	  

This is actually a common fact summarized by the following proposition.

	                                                                                                       ## 
	  

So, for us to see inconsistency from these fields, we need at least two of them.

	                                                                                                       ## 
	  

While we are at the topic, I'd like to use the Ramification Theorems to handle an issue in NDB Cluster described by the following example.

	             T1                                                                        T2

        start transaction;

        insert into t (id2, id3, value) values(1,1,1);

                                                                                start transaction;  

                                                                                insert into t (id2, id3, value) values(2,2,2);

                                                                                rollback;

        insert into t (id2, id3, value) values(3,3,3);

        commit;                                                                                             
	  

	                                                                                                       ## 
	  

There are two ways to remedy this. The first way is NOT to use an auto_increment column and use an explicit counter like d_next_o_id in TPCC to assign values to id1. Every time we request for id1, the counter is updated with a long lock and the interleaved execution in the last example is impossible to show up at all. The second way is to use the Ramification Theorems to control the damage. So if we may design our application so that update of any other field is not affected by id1, this inconsistency will remain visible only at column id1.

Another application of the Ramification Theorems is the following: Suppose the set of transactions and the set of columns in an application are represented as T and C respectively. And they can be segmented into two parts, T1 with C1 and T2 with C2, such that T1 operates on C1 only and T2 operates on C2 only. Then we have Ram(C1) = C1 and Ram(C2) = C2 and the split of T on C1 is T1 and the split of T on C2 is T2. According to the Ramification Theorems, we may place prevention measures on T1 and T2 separately to guarantee the consistency of C1 and C2 respectively. This way we've partitioned both the database and the application into two parts and can deal with consistency one by one(different Ramification Theorems may be applied to C1 and C2 separately). The trivial case is when C1 and C2 correspond to separate set of tables and this result coincides with common sense.

Let's extend this idea to look at a not so trivial case: proving that the new tables and access to them provided as prevention measures in Example 16 will not generate new inconsistency into the original application that the original prevention measures can't handle. The newly added tables are like t_s_n, t_o_n and t_d1_n in Example 16 and they are accessed by statements like

	  select * from t_s_n where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id lock in share mode;                                                                                                  
	

and

	  select * from t_s_n where s_i_id=:ol_i_id and s_w_id=:ol_supply_w_id for update;                                                                                                   
	

and the insertion and deletion of rows in these new tables. The selects are added to the original transactions as prevention measure of course. For the insertion and deletion statements, we have the following observation: TPCC is just the skeleton of a warehousing application and we don't have a transaction to insert rows into, for example, the stock table; in a real world application, transaction like this must be present and we place the corresponding insertion into t_s_n in it; the same story is for deletion of rows in t_s_n. We call the resulting set of transactions T'.

In the case the original tables don't contain any dateime fields, we denote them as C1 and these new tables as C2. Then we have Ram(C2)=C2(calculated with T') since there is no sequence as described in Lemma 1 that connects a read in C2 and a write in C1. The split of T' on C1 is just the original application represented by the set of transactions T. Applying the Ramification Theorem to C2, we may place prevention measures on the original application to guarantee consistency of C1, which are the original tables. This means the extra conflicts introduced by SQL statements to the new tables will not generate new inconsistency into the original application that the original prevention measures can't handle. In the case the original tables contain dateime fields, named the set to be B as usual, we need to set C1=Ram(B)'(calculated with T) and the rest of the columns to be C2, and then we still have Ram(C2)=C2(calculated with T'). The rest of the argument remains the same.

The fields in C2 are insignificant in both cases. The fact that rows in the newly added tables are never updated between insertion and deletion implies they will function as expected even if inconsistencies can be observed in C2. This argument can be applied to any application that implements the method demonstrated in Example 16 and we've just proved the following

This same paradigm can be used to help design a database application sometimes. Besides those tables accessed by clients, we often maintain tables that are accessed by the staff only, which may contain metadata, data for internal uses, etc.. For example, if in the TPCC application orders can be deleted by clients, sometimes we may implement it like this: instead of deleting an order directly from the orders table, we may insert the order into a newly added deleted_orders table without changing the orders table so that the orders table always contains a record for all the orders that have been placed. We name the tables accessed by clients to be C1, the set of transactions for clients to access C1 to be T1, and the tables accessed only by the staff to be C2, which includes the deleted_orders table. If we design the application so that the set of transactions used by the staff, named T2, only accesses C2 and the insertions to the deleted_orders table are the only way for T1 to access C2, then we have Ram(C2)=C2(calculated with T1 + T2) and the split of T1 + T2 to C1 only consists of the statements to access C1 in T1. Hence only placing prevention measures on these statements will suffice to guarantee consistency on C1 by the Ramification Theorems. And the significance of this design is that no matter how the staff use T2 to access C2, it won't affect consistency of C1 and this is desirable in many cases. In the case C1 contains a set B of datetime fields, a similar argument applies by replacing C2 with C2 + Ram(B) and C1 with Ram(B)'. There is yet another advantage of implementing the deletion this way: NOT deleting an order in C1 potentially incurs less conflicts in the split of T1 + T2 to C1. The insertion into C2 potentially incurs more conflicts between statements to access C2, but C2 is for the staff only and they should be more experienced to handle inconsistencies there.

3.5 Joins, sub-queries and unions

In the TPCC application we break down joins into semantically equivalent SQL statements involving only one table. Now let's take a closer look at them.

The first one is in the new-order transaction as follows:

      select c_discount, c_last, c_credit, w_tax from customer, warehouse
         where w_id=:w_id and c_w_id=w_id and c_d_id=:d_id and c_id=:c_id;.                                                                                                    
	

We break it down into these two statements:

      select w_tax from warehouse where w_id=:w_id;

      select c_discount, c_last, c_credit from customer 
         where c_w_id=:w_id and c_d_id=: c_d_id and c_id=:c_id;.                                                                                                    
	

After we find out that there is no update on the 'decision set's involved in these two selects, we may safely combine them back into the original join statement. This way we can take advantage of the 'push down join' algorithm of NDB Cluster and achieve higher performance.

The second one is in the stock-level transaction as follows:

      select count(distinct(s_i_id)) from order_line, stock
         where ol_w_id=:w_id and ol_d_id=:d_id and ol_o_id < :o_id
         and ol_o_id>=:o_id-20 and s_w_id=:w_id and s_i_id=ol_i_id and s_quantity < :threshold;.                                                                                                   
	

We break it down into the following statements:

      select distinct(ol_i_id) from order_line 
         where ol_w_id=:w_id and ol_d_id=:d_id and ol_o_id < :o_id and ol_o_id>=:o_id-20;

      for each ol_i_id obtained in the last statement, assuming it is stored in an array cell :ol_i_id[i] {

        //s->n
        select * from t_s_n
           where s_i_id=:ol_i_id[i] and s_w_id=:w_id
           lock in share mode;

        select count(s_i_id) from stock 
           where s_i_id=:ol_i_id[i] and s_w_id=:w_id and s_quantity < :threshold;
      }.                                                                                                    
	

The necessary prevention measure for the update of s_quantity is also included for comparison that follows.

There is an alternative way to break this join down though:

      select s_i_id from stock
         where s_w_id=:w_id and s_quantity < :threshold;

      for each s_i_id obtained in the last statement, assuming it is stored in an array cell :s_i_id[i] {

        select count(distinct(ol_i_id)) from order_line
           where ol_w_id=:w_id and ol_d_id=:d_id and ol_o_id < :o_id
           and ol_o_id>=:o_id-20 and ol_i_id=:s_i_id[i];
      }.                                                                                                     
	

If we were to use granular locking to prevent the possible predicate RW conflict in this case, it will look like the following since the update is upon s_quantity:

      //s->n
      select * from t_s_n
         where s_w_id=:w_id
         lock in share mode;.                                                                                                     
	

This is a much wider granularity than the previous one and it is certainly not desirable. In general, if there is a way to tell which execution plan the system prefers a priori, we may combine the statements back to the original join and move the prevention measure before it since Theorem 2' applies here. In NDB Cluster, for example, one could apply an explain statement to the join and the 'type' output of it should give us a hint how the join is resolved. If MySQL sticks to the inferred execution plan(I am not 100% sure about this though. For example, say, NDB Cluster favors the execution plan starting with join table t1 in general. But what if at run time, join table t2 only contains one row? Does the optimizer change its mind? A smart one probably should), then we have the 'push down join' back.

The key observation here is that prevention measure depends on the execution plan. We will see this more clearly if the update is upon a field involved in the join condition in the second join. For example, what happens if the update is on s_i_id? In the first execution plan, this gives rise to a predicate RW conflict; in the second execution plan, an item RW conflict results. Right now, we use different conflict resolution measures for them. This implies we can't be sure what to use until the execution plan is resolved. In this Ad-Hoc solution in the second tier for NDB Cluster, we have to break the join down into a specific execution plan a priori to be sure about this. This means we can't take advantage of the 'push down join' algorithm of NDB Cluster any more. One possible solution is to give the optimizer a hint about which execution plan to choose(it might require new SQL syntax). With a capable system like NDB Cluster, there is the following technical path: complicated joins usually show up in OLAP loads, not OLTP ones, so it is probably OK to just break down the simple ones as we did without worrying the 'push down join' algorithm; in the case the load shows both OLAP and OLTP characteristics and the OLAP part is a set of selects, we may try to isolate the OLAP part and service it from a replication server, leaving the OLTP part to the primary cluster(we need to pray that a complicated multi-table update or delete would not be present).

In the case of sub-queries and unions(intersects and excepts too), this execution plan issue doesn't show up unless they contain joins in them. Otherwise, for example for a sub-query, MySQL always evaluates it from inside to outside unless it is a correlated one. This makes the conflict analysis easier for them and I won't elaborate. Of course, we always have the option to break it down as a bunch of statements that only involve a single table as we did for joins.

3.6 Durability of consistency

So all of these deal with the consistency problem in NDB Cluster. This means an application with all the necessary protection and prevention measures will leave a database consistent if it started so and the application finishes successfully. What if the system fails when the application is running? Will it leave the database in a consistent state? How about replications? Do they leave the slave in a consistent state in a traditional replication setting? What about in a more advanced active-active setting? What if a network partition strikes and results in a split-brain situation? We'll answer these questions next.

We first consider the situation when a NDB Cluster suffers a system-wide failure and we call the answer to the question 'whether it will have a consistent state after the cluster is recovered from such a catastrophe' durability of consistency. To do that, we need to recap a key concept called 'epoch' in NDB Cluster first.

Every 100 milliseconds or so, a node in the cluster currently 'elected' as DIH Master initiates a two-phase protocol called Global Checkpoint and sends a GCP_PREPARE_REQ signal to all nodes including itself. On receiving a GCP_PREPARE_REQ signal, the node immediately blocks any prepared transactions from 'beginning to commit'. This causes the Transaction Coordinator (TC) blocks in each node to queue transactions which request commit. Already committing transactions can continue to commit, transactions can continue to be prepared and aborted. Once commit is blocked, the node immediately sends a GCP_PREPARE_CONF back to the DIH Master.

When all GCP_PREPARE_CONF signals have been received, the DIH Master immediately sends a GCP_COMMIT signal to all nodes, which unblocks COMMIT request processing and increments the Global Checkpoint Index (Number) (GCI) associated with every transaction which commits after this point. After unblocking Commit initiation, GCP_COMMIT processing then waits until all of the transactions with the previous GCI have completed, before sending GCP_NODEFINISH back to the DIH Master. Once all GCP_NODEFINISH are received, the DIH Master considers the GCP round complete.

The logical time between two consecutive increments of GCI is called an 'epoch' and the commit of a transaction is associated with an 'epoch' uniquely.

NDB Cluster uses neighborhood-Write-Ahead-Logging or nWAL to provide high availability in case of node failure. Namely, a write is written to all replicas in a node group before it is published to an API node, like a mysqld server, as committed. And before a tuple is written, an entry is written in the REDO log to record this write event. The REDO log at each node is flushed to disk every second. In this case a node fails, the NDB kernel just use the rest of the node group to continue providing service since everything there is the same as in the failed one. In the case the whole node group or the whole system fails, the REDO log is used to restore the system to a consistent state. The question is: does the term 'consistent' used by NDB Cluster here mean the same thing as consistency in this article would have implied?

So we examine the conditions in type B of the Serializability Theorem and easily find out every other condition is satisfied except maybe Condition 2'.

Condition 2' requires the write event of a version exists if a read event of it does and precedes it. In the case of a system-wide recovery, the local REDO log in each node is used to recover itself to less than one second before the failure. If this recovery is NOT to an 'epoch' boundary, there is a chance that the aforementioned requirement might fail.

For example, let n1, n2 be two data nodes in the cluster. N1 coordinates transaction T1 and n2 coordinates T2. Say T1 updates a tuple on n1 which is later read by T2 and this reading results in T2's own update. Assuming both T1 and T2 commit and their modifications are recorded in the local REDO logs on n1 and n2 respectively(for instance, T1 and T2 only modify data on their own coordinator). If a system crash strikes after T2's log is flushed, but before T1's is, then in the recovery afterwards results in T1's update is NOT reflected in the database while T2's is. This is actually a violation of Condition 2' since although T2's read is not recorded in the REDO log, it's implied by its update. Inconsistency could arise from this in general. For example, in the case the tuple read by T2 consists of pay grade info of an employee and that info is used by T2 to update the employee's salary, and the consistency is that any employee's salary must be correlated to his/her pay grade, then the failure to log T1's modification could result in a violation of this consistency. This example also provides better understanding of Condition 2': it doesn't just require a write to be present and 'happened before' its read, this relationship must also be captured by the database.

If this recovery is up to an 'epoch' boundary, in other words, the REDO log is played to the largest 'epoch' that survives the crash(the largest 'epoch' that is persisted to disk in the REDO log for every node), we can prove Condition 2' still holds and type B of the Serializability Theorem applies.

We start by proving the following fact: if an update is recorded in a later 'epoch', it can't be read by a transaction committed in the current or an earlier 'epoch'. We prove it by contradiction. Suppose such a read could happen in the current or an earlier 'epoch'. Then this read event 'happened before' the commit of its transaction, which 'happened before' the commit of the updating transaction by the description of the 'epoch' mechanism since it commits in a later 'epoch'. On the other hand, we are in the Read-Committed isolation level of NDB Cluster, so the commit event of the updating transaction 'happened before' the read. This contradiction proves what we have proposed and leads to the following conclusion: a read only reads updates from the current or an earlier 'epoch'. And since the recovery is up to an 'epoch' boundary, everything before that boundary has been recovered, including any update read by a transaction.

According to an article by Frazer Clement which is available here, “epoch boundaries act as markers in the flow of row events generated by each node, which are then used as consistent points to recover to”. This means every node in the system is recovered to the same 'epoch' boundary in a system-wide recovery. Hence the recovered database state is consistent by type B of the Serializability Theorem.

This conclusion remains true if we instead are applying the Generalized Serializability Theorems to our application since it is like applying type B of the Serializability Theorem to the Read-Write transactions, for example. And they are the only transactions that can affect the underlying database. For the Ramification Theorems, the conclusion remains true on Ram(B)'.

For traditional replication where system-wide changes are replicated to a slave using a binlog, the slave always maintains a consistent state by type B of the Serializability Theorem. The reason is that all the changes in an 'epoch' in all the data nodes in the system is grouped into an 'epoch' transaction, which contains all the writes by transactions in this 'epoch', before it's replicated to the slave according to Frazer Clement's article. From the arguments of the last few paragraphs, this guarantees consistency if the replication runs to its end.

Now what happens if the replication process fails in the middle? If only the replica or the replication channel fails, there are relevant solutions in the MySQL documentation and I won't elaborate on those. We only consider the situation where the source cluster also suffers a system-wide failure here. In that case, from the discussion about durability of consistency, we know that we can always recover the source to a consistent state. After syncing that state to the replica, we may resume replication.

In an advanced active-active setting where all clusters in the system are able to handle write loads, conflicts are resolved using mechanisms described in a series of articles by Frazer Clement, which are available here. Conflicts come into two flavors in this context: tuple-based and transaction-based. Tuple-based conflicts arise when the primary and secondary cluster modify the same tuple concurrently. When a cluster replicates the update to its peer, it could result in a different order in either of them. The resolution is to rollback the modification from the secondary cluster and let the one from the primary prevail. But if there are other updates in the transaction encapsulating the update from the secondary, this resolution mechanism could lead to a situation where only part of the transaction is aborted since not every update in it results in a conflict. This abnormally is called 'data-shearing' in Frazer Clement's series and hence transaction-based conflict is introduced to take care of it. Transaction-based conflict can be defined with the following quote from Frazer Clement's article available here:

“Where a row is found to be in-conflict with some replicated row operation, a further replicated row operation on the same row should also be found to be in-conflict, until the conflict condition has been cleared. This property is implicitly implemented in the existing row based conflict detection functions.

When the scope of a conflict is extended to include all row modifications in a transaction, this implies that all following replicated row operations which affect the same rows, must also be in conflict by implication. To avoid row shearing, these implied-in-conflict rows must implicate the other rows in their transactions, and those rows may in-turn implicate other rows. The overall effect is that a single row conflict must cause its transaction, and all dependent transactions to be considered to be in conflict.”

This means if a transaction on the secondary cluster is in-conflict, all transactions in the same 'epoch' with WW conflicts with it are considered to be in-conflict. And recursively every transaction in that 'epoch' is examined to see if they are in-conflict with the previous found ones until the set of found ones doesn't expand any more. This set of transactions are then rolled back and the conflict resolution process moves onto the next 'epoch'.

Notice that this scheme doesn't suffer the same problem as in the durability of consistency case. For example, in the secondary cluster, if again we assume that T1 updates the database in 'epoch' e, T2 read this update and perform its own update in 'epoch' e+1; simultaneously, T1's update causes a conflict with the primary cluster and T1 is rolled back. Then again the database is in an inconsistent state since that update is gone, but its effect(the read and update in T2) is still there and Condition 2' is violated. This, however, only leaves the secondary cluster's state inconsistent transiently because the conflict resolution process will discover that and roll back T2 shortly when it deals with 'epoch' e+1. So the secondary cluster will be eventually consistent with the primary.

In the case the primary cluster suffers a system-wide failure, the secondary will assume its role. In the unlikely case where the secondary cluster also suffers a system-wide failure afterwards, it is subject to the same situation as in durability of consistency.

When a split-brain situation occurs, there is at most one subset of surviving nodes in NDB Cluster that is going to continue to provide service. The criteria for choosing this surviving subset is:

1. It must contain at least one node from each and every node group.
2. It must also contain a pre-selected node called the arbitrator. In other words, only the subset including the arbitrator qualifies.

Remark: If the network topology is special enough so that the arbitrator can be connected to more than one isolated subset, the arbitrator only responds 'yes' to the first subset that queries it. So the uniqueness of the surviving subset is well-defined.

The mechanism in 2 is called arbitration in NDB Cluster and it is described in an article by Frazer Clement discussing the CAP Theorem based issues here. Notice the consistency in the CAP Theorem means 'whether two copies of the same data in a distributed system hold the same value' and hence is very different from the consistency discussed in this article.

When a node or network failure is detected by the NDB kernel, updates are not allowed to commit; on the other hand, between the occurrence of a split-brain situation and its detection, committed updates are only possible in node groups with all replicas to be up; both are true because the replication between nodes in a node group is synchronous. This implies consistency defined in the CAP Theorem is honored before, during and right after a split-brain if a surviving subset will be able to continue to provide service. After that, this consistency is only honored in the surviving subset until the partition is resolved.

This also implies the consistency discussed in this article is honored since Condition 2' is never violated in the whole process. So in the case a surviving subset exists, after network partition is resolved and other nodes are brought back online, these nodes will be synchronized with the surviving subset before they are serving requests again. Whether this is done during the application is executing or after it's finished execution, it won't affect consistency as long as it does finish.

On the other hand, if no such subset survives the partition, the cluster will shutdown and a restart is needed after network partition is resolved. This behaves exactly like the durability of consistency case.

In the traditional replication scenario, if a split-brain situation occurs in the slave, there are two cases. Case 1: there is a surviving subset that can continue to provide service; we only need to resolve the partition and sync the rest of the slave with the surviving subset. Case 2: there is no surviving subset that can continue to provide service; after partition is resolved, the slave need to be initialized and replication need to be set up again with the master. There is a slight complication in Case 1, if the SQL node used for replication is not attached to the surviving subset. We need to stop the service on the slave first and then set up replication with a SQL node in the surviving subset or treat it as in Case 2 and set that up after partition is resolved before service can be restored.

If, on the other hand, a split-brain situation occurs in the master in the traditional replication scenario, there are also two cases. Case 1: there is a surviving subset that can continue to provide service; we only need to resolve the partition and sync the rest of the master with the surviving subset. Case 2: there is no surviving subset that can continue to provide service; in this case, we must restart the master after partition is resolved and consistency will be preserved as in the durability of consistency case. In Case 1, the complication that SQL node used for replication is not attached to the surviving subset must also be addressed. Namely, we need to stop the service on the slave first and then set up replication with a SQL node in the surviving subset before service can be restored.

In an active-active replication deployment, in the case the primary cluster suffers a partition, there are two cases. Case 1: if there is a surviving subset in the primary so that the SQL node for replication is also included, the replication will continue and we only need to resolve the partition and reinstate the primary by syncing the rest of it with the surviving subset. Case 2: otherwise, the secondary will assume its role until partition in the primary is resolved and it is synced with the secondary. In the case the secondary cluster suffers a partition, the situation is similar.

One interesting question is: is there a system that is consistent, but in which durability of consistency could be violated? I don't have a database example for that. But the following incident that resulted in Raimondo, US secretary of commerce's email account being hacked certainly smells like it. The key piece of info there is “The crash dumps, which redact sensitive information, should not include the signing key. In this case, a race condition allowed the key to be present in the crash dump (this issue has been corrected).”. The consumer signing system in question is not a fault-tolerant system. Faults show up in a fault-tolerant system as inconveniences, not mishaps.

3.7 Application to standalone systems

Although the method developed in this article emphasizes distributed database systems, it is backward compatible with and can be applied to standalone systems. In this subsection, we'll explore how to do that in various standalone systems, starting with MySQL InnoDB. If other than consistency, the expected performance profile for your application is at most a few thousand tps, you are at the right place.

	                                                                                                       ## 
	      

	                                                                                                       ## 
	      

	                                                                                                       ## 
	      

4.1 Isolation levels for a field-based database system

In particular, this definition implies transactions updating different fields in the same tuple can be executed concurrently. And it also imposes a total order on field versions of a field or 'decision set' versions of a 'decision set'. For a distributed database system, since we can't synchronize clocks to 100% accuracy by [La 78], we must provide a new way to generate the timestamps. For example, Google's Spanner comes with a way to implement Snapshot Isolation without synchronizing clocks accurately, which is described in Appendix D. More example will be given in next section.

Also to facilitate such a Snapshot Isolation implementation for a field-based system, we need to associate each field version with a timestamp, instead of one for each tuple version. This will increase storage footprint in general. However, there are usually large amount of memory and disk space in modern servers. And a distributed database system can also be scaled horizontally. This should not be a problem.

This definition applies to both standalone and distributed database since the term 'latest' refers to a local clock. And a field level capable locking system is needed for this Read-Committed isolation level.

Before defining a new Repeatable-Read isolation level, we need to take a closer look at MySQL InnoDB's.

	             T1                                                                  T2

        start transaction;

        select * from t;

                                                                            start transaction;

                                                                            update t set id2=1 where id1=1;

                                                                            update t set id2=4 where id1=3;

                                                                            commit;

        update t set id3=3 where id1=1;

        select * from t;

        commit;                                                                                
	  

	                                                                                            (To be continued ...)  
	  

This behavior of MySQL InnoDB is reasonable since its Repeatable-Read isolation level is a tuple-based system and the tuple version established by T1 is after that of T2's, so T1 should be able to see what the previous version is. With a field-based system where the concept of tuple version is absent, the story is very different as we will see shortly.

Remark: If we place a statement 'select * from t;' into T1 right after T2's commit and before T1's update, it will return the two original tuples too.

For a distributed database system, since we can't synchronize clocks to 100% accuracy, we must provide a new way to generate the timestamps. Also as in the Snapshot Isolation case, we need to associate each field with a timestamp.

These three field-based, newly defined isolation levels are all variants of the generic Read-Committed isolation level in the SQL standard since they all proscribe the following two anomalies:

      P0: w1[x] … w2[x] … (c1 or a1),

      P1: w1[x] … r2[x] … (c1 or a1), a for abort and c for commit here.                                                                                      																								
    

The three field-based, newly defined isolation levels mainly concern P1, but P0 is implicit according to Adya's paper ([La 78]). Of course, we should supply a version of P1 for predicate reads here so that it becomes:

      P1: w1[x] … r2[x] … (c1 or a1),

          w1[x] … r2(P: Vset(P)) … (c1 or a1), a for abort and c for commit here.
    

4.2 Demonstration of type C of the Serializability Theorem under the newly defined isolation levels

Since type C of the Serializability Theorem is based on the generic Read-Committed isolation level with conflicts interpreted at the field level, this means Condition 5 & 6 are both proscribed by the three field-based, newly defined isolation levels. Conditions 1, 2', 3' & 4 are just common sense. Also we require the write events of a field to form a serial order that is consistent with the 'happened before' partial order in type C of the Serializability Theorem and it is certainly satisfied by the definitions of the three isolation levels. Hence we may apply type C of the Serializability Theorem to the three field-based, newly defined isolation levels as long as we may prevent conflict loops to form.

	                 T1                                                          T2

        start transaction;
 
        select * from t;

                                                                        start transaction;

                                                                        update t set id2=1 where id1=1;

                                                                        update t set id2=4 where id1=3;

                                                                        commit;

        update t set id2=id2+1 where id1=1;

        select * from t;

        commit;                                                                                   
	  

	                                                                                            (To be continued ...)  
	  

4.3 TypeD of the Serializability Theorem

To support symmetric access via multiple attributes and get rid of Example 0, we need a new type of Serializability Theorem other than type C, namely, type D of the Serializability Theorem in which the 'decision set' could be of more than one field. To do that, we are going to capture a 'decision set' version concept out of the system for type C where only field versions are present. In other words, we are assuming a system that offers a generic Read-Committed isolation level and only field versions are available in the following discussion. This is very different from a system with the concept of tuple version where we may simply derive a 'decision set' version from a tuple version.

We will first use a trick to capture the 'decision set' version concept and then generalize it for type D of the Serializability Theorem. The trick is to pick a field in the 'decision set' and require that each update of the 'decision set' also modifies this specific field. The modifications of the specific field naturally impose a total order on its field versions and we hope that this total order will lead to a total order among the 'decision set' versions to be defined. It turns out this goal can be achieved as follows: because the total order of the specific field versions is generated by a sequence of serial write events as we mentioned after the definition of the 'happened before' partial order, we may require the following condition to hold.

Suppose the 'decision set' consist of fields {a, b, c}, c being the specific field. Transaction T1 updates all of them while T2 only updates b & c such that the order of update of c is T2 before T1. First, let's observe that the update of b by T2 'happened before' that of T1 too. Proscribing P0 implies, in any reasonable implementation, the write by the second transaction must wait until after the commit of the first one since otherwise if the first transaction, for example, got context-switched out, P0 might result. So if the writing of b by T1 'happened before' that of T2, the commit event of T1 'happened before' that of T2 too since it 'happened before' the write of b by T2. But we also know that it should be the other way around from the writing of c. This is a contradiction and the observation is correct.

Then the update of b in T1 is confined between write events of c since: the update of b in T1 'happened before' the update of c in T1 is asserted by Condition **; the update of c in T2 'happened before' the commit of T2, which in turn 'happened before' the update of b in T1 because P0 is proscribed. The update of a in T1, however, can overlap with that of c in T2. This guarantees a new 'decision set' version is generated whenever the write event of the specific field ends: for fields updated in the current transaction, the versions just generated are used; for fields not updated, the latest generated versions or those from original database state are used. With this definition of a 'decision set' version, the writes on the specific field give rise to a 'decision set' version system and a total order can be imposed on the 'decision set' versions so that total order of its derived field versions on that specific field coincides with the original total order of the specific field versions. It is also easy to see that the order of 'decision set' versions coincides with that of other fields since they are both consistent with the order of the specific field.

And what if two multi-field 'decision set's overlap? Do they introduce the same 'derived' versions on the set of overlapping fields? It turns out they do. The observation is that for each 'derived' version, the overlapping set must be modified and this implies both 'derived' version sets coincide. So the only problem is these two 'derived' version sets may not have the same total order imposed. But that is impossible since by the previous analysis, the total order imposed on both 'derived' version sets must coincide with the 'happened before' partial order and hence must be the same. A similar argument can be used to show that the 'derived' field versions from these 'derived' overlapping set versions coincide with the original field versions. So the definition of this 'decision set' version system is sound.

Remark: In Condition **, we use the completion of update of the specific field to signify the establishment of a new 'decision set' version. The fact that no transaction can access the 'decision set' version before that doesn't mean they have to be accessed right after. For example, if we were to implement such a field-based system for NDB Cluster, this completion of update can be chosen to be when the primary replica gives up the field locks on a tuple. Then the 'decision set' version generated can't be updated until the corresponding locks are given up in the secondary replica. On the other hand, if we were to implement it for MySQL InnoDB, we may choose this completion of update to be when field locks are released and the 'decision set' version will be available for both read and update immediately. So Condition ** is defined in such a way that it will allow different implementations.

In the newly defined Read-Committed isolation level, a write event can be represented by the acquisition and release of the field lock. To satisfy Condition **, we may acquire field locks for all the fields written in an order so that unnecessary dead locks wouldn't arise, update the fields, and then release the locks simultaneously; or we may request a lock that covers all the fields written in the 'decision set' if the field level capable locking system supports it and perform the updates while the lock is held. Notice in the former case, it is usually not possible to acquire all the field locks simultaneously since some of them may require lock wait, but it is in general possible to release them at the same time. In the newly defined Snapshot Isolation isolation level, writes on the specific field clearly generate 'decision set' versions since corresponding transactions are non-concurrent and it also satisfies Condition **. In the newly defined Repeatable-Read isolation level, the 'decision set' version is generated as in the Read-Committed isolation level and Condition ** is satisfied for the same reason as in the Read-Committed isolation level case. Notice in both Snapshot Isolation and Repeatable-Read case, we may use the timestamp for the specific field version as that of the corresponding 'decision set' version.

If we try to use the 'decision set' version we've just defined for a predicate read, we must make sure it is the 'decision set' version that we use, not an arbitrary combination of field versions in the 'decision set' since we are now in a field-based system. In the newly defined Read-Committed isolation level, this can be done by acquiring short read lock on the specific field before the predicate read of a tuple and make sure the matching process happens in the duration the lock is held. Since updates for the fields in the 'decision set' only happen after the specific field is locked, the short read lock guarantees the reading of a 'decision set' version. In the case an initial internal version for the 'decision set' is read, the write lock acquired on the specific field before the read should guarantee the reading of a 'decision set' version. In the newly defined Snapshot Isolation isolation level, the predicate read returns all the latest field versions in the 'decision set' that are before the starting time of the transaction and the fact that these field versions constitute the latest 'decision set' version is clear from the way 'decision set' versions are generated. In the case an initial internal version for the 'decision set' is read, the situation is the same. In the newly defined Repeatable-Read isolation level, as Example 25 indicates, different scenarios arise. If the 'decision set' of the predicate read is NOT altered in the transaction or the predicate read happens before the alternation, it behaves the same as in the Snapshot Isolation isolation level case. Namely, it doesn't see any 'decision set' version established after the starting time of the transaction. Otherwise, if the predicate read happens after the alternation, it behaves the same as in the Read-Committed isolation level case. Namely, it sees 'decision set' versions established after the starting time of the transaction. In either case, we may have it return a 'decision set' version for the predicate read.

	                                                                                            (To be continued ...)  
	  

In general, we should impose the following condition when transactions interact with the 'decision set' version system.

From the discussion above, we know that the three newly defined isolation levels all satisfy Condition ***.

Remark: If we were to implement a field-based system for NDB Cluster, Condition *** can be satisfied easily since it is just an implementation of the newly defined Read-Committed isolation level.

With an initial internal 'decision set' version defined, we may apply updates of the 'decision set' to it successively and create an internal 'decision set' version system when a transaction need to modify this specific 'decision set'. And that is the last piece of the puzzle for presenting the following

	                                                                                                       ## 
	  

In general, we need a systematic way to define a 'decision set' version system so that the following requirements are met for type D of the Serializability Theorem.

1. A way to define the system.
2. A way to read the 'decision set' versions in a predicate read without mistakenly using an arbitrary set of field versions in the 'decision set' instead, so that predicate-based conflicts can be defined.
3. A way to read the 'decision set' versions for the initial internal version without mistakenly using an arbitrary set of field versions in the 'decision set' instead, so that successive internal versions and eventually the committed 'decision set' version can be defined.
4. And also a way to introduce item-based conflicts between reads and writes of 'decision set' versions so that we can reason about conflicts with them.

Remark: Again we define in requirement 4 a third kind of conflict other than the item-based conflicts between fields and predicate-based conflicts as in type B of the Serializability Theorem. In type A, this is buried under the concept of tuple versions and in type C and the type with 'decision set' versions induced by a specific field we just presented it is opaqued by conflicts between field versions and the specific field versions. In type B, we've tried to isolate it from the tuple versions. And now in type D we've made it stand out more clearly since we don't have tuple versions in the way any more. This is the exact ingredient we need to fence off Example 0 since now updating different fields in a 'decision set' will result in a conflict.

Of course, it also requires a way to identify which field version to read after a predicate read if item reads follow as in type C. All these lead to a new type of the Serializability Theorem as follows.

	                                                                                                       ## 
	  

This is not a surprise since type D of the Theorem is just a generalization that we hoped the Ad-Hoc one with a specific field would have led to. Hence what is remaining is to define a 'decision set' version system for all three newly defined isolation levels so that they satisfy all four requirements so that we can apply the theorem to them.

In the newly defined Snapshot Isolation isolation level, we may require updates to the same 'decision set' of a tuple to be conflicting. This way transactions with these conflicting updates can NOT to be concurrent. Hence the 'decision set' versions are defined with the temporal order without needing the specific field and can be labeled with a timestamp for each of them. The way to read it in a predicate read or as an initial internal version is timestamp based, just as reading a tuple in a tuple based system. Starting with the initial internal version, we can write all the internal versions and eventually a committed version is generated and timestamped if the transaction commits. And the item-based conflicts between these writes and reads of the 'decision set' versions are defined like in a tuple based system. Again we can argue that consistency issues related to 'derived' versions will not arise as in the case where 'decision set' versions are defined with the help of a specific field.

In the newly defined Read-Committed isolation level, things are more complicated since a field level capable locking system has to be in place and there could be different implementations. While we are at the subject, I'd like to discuss a couple of alternatives and hope it would help pave the way to such an implementation in the future.

Alternative one: In the field level capable locking system, we may acquire a lock that covers exactly the 'decision set'. It is like a tuple lock, except that it might cover less fields. And we call it a 'decision lock'.

Alternative two: In the field level capable locking system, we may only acquire field locks to cover all the fields of the 'decision set'.

For alternative one, we consider two strategies. Let's say the 'decision set' in concern consists of fields {a, b, c, d, e} and we call it I. And there is also another 'decision set' overlapping with it with fields {d, e, f} and we call it II.

In strategy one, when 'decision set' I is updated, if d or e is also updated, we acquire a long 'decision write lock' that covers all fields a-f; if neither of d or e is updated, we acquire a long 'decision write lock' that covers fields a-e. Once the lock is acquired, 'decision set' I is updated('decision set' II is also updated if d or e is updated) and a new version of I is established when the lock is released at transaction commit. This guarantees serial updates of 'decision set' I and the versions for 'decision set' I are well-defined. The most recently generated 'decision set' versions are used in predicate reads or the read for the initial internal version by requesting a short 'decision read lock' or a long 'decision write lock' respectively. Once the initial internal version is read, updates are applied to it in order to generate the internal versions and eventually lead to the committed version of a 'decision set'. And the way to recognize the item-based conflicts between these reads and writes of a 'decision set' is the same as in a tuple system case, even if writes don't necessarily share common fields.

In strategy two, when 'decision set' I is updated, if d or e is also updated, we acquire a long 'decision write lock' that covers all fields a-e and a long 'decision write lock' that covers fields d-f(these two locks don't conflict with each other if they are requested by the same transaction); if neither d or e is updated, we only acquire a long 'decision write lock' that covers fields a-e. In either case, 'decision set' I is only updated after all the necessary locks are acquired. In this strategy, a 'decision set' version for I is well-defined since between the time the first lock is acquired and the last lock is released, no other transaction can acquire the long 'decision write lock' that covers a-e. And because an update only happens when 'decision set' I is covered with this 'decision write lock', it implies updates to 'decision set' I are serial. Notice the time between the first lock is acquired and the last lock is released for different transactions could overlap with each other. For example, if there was also a 'decision set' III consisting of fields a, b & c and the order we were acquiring and releasing these locks in a transaction was III, I, II, then there was a chance that one such transaction was releasing the lock for II and another such transaction was already acquiring the lock for III since they don't conflict with each other. Hence comparing with strategy one, the order of 'decision set' versions is a bit fuzzier, but it is still NOT ambiguous. The reading, writing and the item-based conflicts part of a 'decision set' will be the same as in strategy one. Of course to make the whole thing work, we probably should impose a global order to eliminate dead locks. But we are not going to elaborate on that here since strategy one is certainly a simpler choice and the main purpose of strategy two is to induce the following alternative two.

A similar argument as in the Serializability Theorem with 'decision set' versions induced by a specific field case can be used to show that 'derived' versions from common fields of two overlapping 'decision set's will agree. So the definition of 'decision set' version system is sound.

In the newly defined Repeatable-Read isolation level, everything is similar to the Read-Committed isolation level case except that we need to assign a timestamp to each 'decision set' version to facilitate reads in a repeatable-read manner. Since things are a little bit fuzzy in strategy two of alternative one and alternative two, we need to be careful about which point of time we choose as the timestamp. The point in time when the first of acquired locks is released could to be one such choice. Then all four requirements for a 'decision set' version system will be satisfied and hence we've proved the following

	                                                                                            ## 
	  

All the Generalized Serializability Theorems and Ramification theorems are also true for type D's setting. The proofs are similar to those for Generalized Serializability Theorems and Ramification theorems for type B, except this time the 'decision set' is not necessarily derived.

4.4 Relations between different types of the Serializability Theorem

Let's take a look at the relations between type D and the other three types of the Serializability Theorem. First type C is a special case of type D where its 'decision set's happen to contain just one field. Type B, of course, is a special case of type D where the 'decision set' version system is derived. So proving type D of the Serializability Theorem also proves type C and type B of it.

We claim we can use type D of the Serializability Theorem to prove type A of it. The 'if' part is the same as before and the arguments for the 'only if' part are as follows.

	                                                                                                       ## 
	  

The choice of topological sort based on DSG(A) is important in this proof and we might be allured to choose one based on DSG(D). The following simple example tells us why we should not.

	                                                                                            ## 
	  

From the relations between the four types of the Serializability Theorem, we know that type D is the most generic one. This is not a surprise since in type D we look at things at the field level – the finest granularity that is possible. It should be able to explain types that are of coarser granularity.

4.5 Type H of the Serializability Theorem

Although finer control of conflicts can be achieved in type D of the Serializability Theorem, more resources and overhead may be necessary. For example, in a typical lock manager, usually one control block is necessary for each lock. Hence if we implement locks at the field level, depending on the implementation, a lot of extra memories and maintenance might be needed. On the other hand, in type A of the Theorem, unnecessary conflicts are introduced while resources and overhead might only leave a more reasonable footprint. So is there a way to combine the merits of both approaches? The previous proof actually suggests an affirmative answer for applications accessing more than one table.

The key observation is that any conflict between a pair of transactions is actually interpreted in a specific table. This means we only need information like which tuple or field a transaction writes, or the range of tuples that a predicate read covers in a specific table to decide whether a conflict exists. Hence we may interpret these conflicts on a per table basis. In other words, we may interpret conflicts at tuple level for one table and do that at field level for another. This gives rise to the first hybrid Serializability Theorem as follows:

	                                                                                                       ## 
	  

So the next question naturally is: on what tables should we use tuple-based/field-based conflict resolution? We introduce a concept of 'logical group' within a tuple to answer it.

Some fields in a tuple must co-variate with others and they should be put into a logical group. For example, consider the customer table in the TPCC application. There is a group of fields related to the customer's address info and there is also a group of fields related to the customer's credit info. These two can be viewed as two different logical groups. That is because in a real world warehouse application, after the tuple's initial insertion, these two groups are probably modified by two independent transactions. If we use field-based conflict resolution on the customer table, these two transactions don't have to conflict with each other at the lock level in a lock-based system; on the other hand if tuple-based conflict resolution is used, they are guaranteed to conflict with each other at the lock level.

So the answer to our question is: if there are more than one logical groups in a table, consider using field-based conflict resolution; on the other hand, if there is only one logical group within a table, tuple-based conflict resolution is enough. Notice that if fields in a key in a table are only used in a predicate read, they don't constitute a logical group, otherwise they do. Also, logical groups are application dependent. For example, if TPCC turns out to be a warehouse application for a postal service, the zip code field and other fields in the address might be modified independently by two transactions since the postal service might change the zip code for a specific address and they constitute two logical groups in that case.

Now that type D of the Serializability Theorem allows us to discuss consistency fully down to the field level, does it have any advantage over its peer, say, type B of it? The discussion about logical groups certainly gives us an affirmative answer. In fact, there is one such example in the TPCC application: in the payment transaction, the field D_YTD is updated while in the new-order transaction, the field D_NEXT_O_ID is updated too; but there is no consistency related conflicts between these two transactions and these two updates don't introduce lock waits in the type D setting. A similar example can be given in the case that a read and a write on different fields are from different transactions as well.

5.1 Making the generalization generic

The following discussion is basically that of [Fe 05] between Remark 2.1 and Theorem 2.1, with two major differences.

The first one is that we assume the distributed clock, whose reading designated as t.e for event e, satisfies the following:

        Clock Condition: e “happened before” f => t.e < t.f,                                                                                        
	

where the inequality is from a total order.

Remark: I came across the Clock Condition in Lamport's classic paper ([La 78]) about distributed clock. It was for the logical clock defined there only. I extend the idea here for a generic distributed clock. In the Clock Condition, we use “happened before” instead of 'happened before' as the rest of this article does. This “happened before” partial order is for events designated as points in time, not as an interval as in the 'happened before' partial order, so that we may associate a timestamp to each event. This way of capturing the temporal order in a reference frame is discussed in detail in Appendix B and the subsection about CockroachDB in Appendix D: events inside a thread are represented as a sequence of points in time and two events connected by an inter-thread message is interpreted as a type II edge. And this potential mixed usage of “happened before” and 'happened before' is OK since 'happened before' is only used in type D of the Serializability Theorem to assert that a non-serializable history contains a cycle in the proof of Theorem 2.1 below. After that, “happened before” is used for the rest of the arguments.

The second one is that we assume t.e could be equal to t.f for two different events e and f. In a standalone system, we can usually arrange the timestamp service to return an increasing sequence so that t.e is not equal to t.f, if e and f are not the same. In a distributed system, however, the distributed clock usually can't fulfill such a requirement. We must cope with it.

With the second assumption, we need to define two concurrent transactions T1 and T2 to be: timestamp of T1's start <= timestamp of T2's commit and timestamp of T1's commit >= timestamp of T2's start; in the case the timestamp service returns an increasing sequence, it is always safe to use the strict inequality signs '<' and '>'.

For a WR conflict from T1 to T2, we know that timestamp of T1's commit < timestamp of T2's start from the definition of Snapshot Isolation. So if there is a WR conflict between T1 and T2, timestamp of T1's start < timestamp of T1's commit(Clock Condition) < timestamp of T2's start < timestamp of T2's commit(Clock Condition).

If the conflict between T1 and T2 is a WW one, by First-Committer-Wins rule or First-Updater-Wins rule, we have timestamp of T1's start < timestamp of T1's commit(Clock Condition) < timestamp of T2's start < timestamp of T2's commit(Clock Condition). Notice the second inequality can't be the other way around(timestamp of T2's commit < timestamp of T1's start) since then we would have the following timestamp loop: timestamp of T2's commit < timestamp of T1's start < timestamp of T1's conflicting write < timestamp of T2's conflicting write < timestamp of T2's commit by the Clock Condition.

If there is a RW conflict between T1 and T2, it is possible that T1 completely precedes T2 or for T1 and T2 to be concurrent since T2 completely precedes T1 would result in T1 being able to read T2's write, which contradicts with the definition of a RW conflict. But in either case, we have timestamp of T1's start <= timestamp of T2's commit. We've just proved the following Lemma 2.2 in [Fe 05] with a distributed clock satisfying the Clock Condition.

From the discussion above, the following Lemma 2.3 in [Fe 05] is also clear.

Now we are ready for the proof of the major theorem in [Fe 05], with a distributed clock satisfying the Clock Condition this time.

	                                                                                                       ## 
	  

So as long as a distributed clock satisfies the Clock Condition, we may apply Theorem 2.1 in [Fe 05] to an application executed under the Snapshot Isolation defined with this distributed clock. In particular, this implies we may apply Cahill's work ([Ca 08]) to it to achieve serializability.

5.2 List of distributed clocks satisfying the Clock Condition

We'll discuss a few distributed clock candidates satisfying the Clock Condition here, hoping that it will help you choose one for your specific implementation of Serializable Snapshot Isolation in a distributed database system.

        Initially l.j := 0

        Send or local event
          l.j := l.j + 1
          Timestamp with l.j

        Receive event of message m
          l.j := max(l.j+1, l.m+1)
          Timestamp with l.j, c.j
                                                                                               
	  

5.3 Alternatives for a field-based Snapshot Isolation implementation

In Section 4, we've shown that type D of the Serializability Theorem can be applied to a field-based Snapshot Isolation level to conclude serializability if an application is free of conflict loops. Let's see how such a field-based Snapshot Isolation level can be implemented here. We'll explore two alternatives next. The first one will be designed so that the cost of adapting a current tuple-based Snapshot Isolation implementation to become a field-based one is minimized. The second one is optimized so that storage footprint of the database is minimized.

        key_version1 → value(tuple version)

        key_version2 → value(field versions)

        key_version3 → value(field versions)

        ...... 
	  

5.4 Primary algorithm for the generalization

The consecutive RW conflicts between concurrent transactions in Theorem 2.1 form the so-called 'dangerous structure' in the nomenclature of Fekete's paper ([Fe 05]). Cahill's team then find a book-keeping way to keep track of these 'dangerous structure's when an application is executed under Snapshot Isolation in [Ca 08]. For every transaction T in the application, two boolean variables T.inConflict and T.outConflict are set respectively if a RW conflict is coming into or going out from T. They are both stored in the transaction record of T. Also, a new SIREAD lock is introduced to capture a RW conflict with a write lock for the First-Updater-Wins rule. No extra blocking is introduced though, since a SIREAD lock is more like a flag to indicate that a read has happened. A SIREAD lock is placed when an item read or a predicate read starts. The following is the primary algorithm in [Ca 08]:

      modified begin(T):

          existing SI code for begin(T)
          set T.inConflict = T.outConflict = false

      modified read(T, x):

          get lock(key=x, owner=T, mode=SIREAD)
          if there is a WRITE lock(wl) on x
              set wl.owner.inConflict = true
              set T.outConflict = true

          existing SI code for read(T, x)

          for each version (xNew) of x
          that is newer than what T reads:
              if xNew.creator is committed
                  and xNew.creator.outConflict:
                      abort(T)
                      return UNSAFE_ERROR
              set xNew.creator.inConflict = true
              set T.outConflict = true

      modified write(T, x, xNew):

          get lock(key=x, locker=T, mode=WRITE)

          if there is a SIREAD lock(rl) on x
              with rl.owner is running or commit(rl.owner) > begin(T):
                  if rl.owner is committed
                      and rl.owner.inConflict:
                          abort(T)
                          return UNSAFE_ERROR
                  set rl.owner.outConflict = true
                  set T.inConflict = true

          existing SI code for write(T, x, xNew)
          # do not get write lock again

      modified commit(T):

          if T.inConflict and outConflict:
              abort(T)
              return UNSAFE_ERROR

          existing SI code for commit(T)
          # release WRITE locks held by T
          # but do not release SIREAD locks  
	

Remark: In this section, instead of First-Updater-Wins rule, we use First-Committer-Wins rule to resolve write conflicts between concurrent transactions. Such an implementation doesn't require long locks. Instead, we'll develop a flag system, which is discussed in detail in next sub-section, to capture RW conflicts. In a flag system, we use a read flag for a SIREAD lock and a write flag for a write lock. But in this sub-section, let's still hold on to the concepts of SIREAD lock and write lock so that it doesn't deviate too much from Cahill's work.

In the 'modified read' part of the algorithm, we can see that whenever there is a version written after the read, it is signified as causing a RW conflict upon item x. So Cahill's team follows strictly what [Be 87] is interpreting RW conflicts: a read conflicts with a write on the same data item, even if there are a thousand other writes between them. This is absolutely unnecessary of course, according to type D of the Serializability Theorem in this article. Only the version written immediately after matters. So we can optimize the 'modified read' part as follows:

	  modified read(T, x) for item RW conflicts:

          if a version (xNew) of x right after what T intends to read exists:
              if xNew.creator is committed
                  and xNew.creator.outConflict:
                      abort(T)
                      return UNSAFE_ERROR
              set xNew.creator.inConflict = true
              set T.outConflict = true
          else:
              get lock(key=x, owner=T, mode=SIREAD)
              if there is a WRITE lock(wl) on x
                  set wl.owner.inConflict = true
                  set T.outConflict = true

          existing SI code for read(T, x)
	

So this is the case for item RW conflicts. For predicate RW conflicts, we can't push down the acquisition of a SIREAD lock as in the optimized 'modified read' for the case of item RW conflicts since a granular lock usually covers a range and after the identification of an IN or OUT operation that is causing a conflict, there might be other IN or OUT operations causing conflicts on other tuples in that range. The optimized 'modified read' for predicate RW conflicts in general is given as follows:

	  modified read(T, R) for predicate RW conflicts:

          get lock(range=R, owner=T, mode=SIREAD)
          if there is a WRITE lock(wl) on x in R
          generating the first version of match change
              set wl.owner.inConflict = true
              set T.outConflict = true

          existing SI code for read(T, R)

          for each version (xNew) of x in R
          that is the first version of match change after what T reads:
              if xNew.creator is committed
                  and xNew.creator.outConflict:
                      abort(T)
                      return UNSAFE_ERROR
              set xNew.creator.inConflict = true
              set T.outConflict = true
	

Here R is a range that covers the predicate read. In pessimistic technology, (multi-)granularity locking is usually employed to fence off conflicting writes to a predicate read. In MySQL InnoDB, R could be just what a bunch of 'gap lock's cover; in the serializability implementation I've developed for NDB Cluster, R could be just what a regular lock in the higher hierarchy covers if the application is well-structured; in TiDB, R could be multiple Regions where the predicate read happens; in Cahill's prototype of Serializable Snapshot Isolation deployed in Berkeley DB, R could be just what a bunch of pages span. Optimistic technology may not employ locks, but the concept of range R remains the same.

In a distributed database system, components of R may scatter into different nodes. We probably should replace R in the algorithm with its local component so that decisions can be made locally to boost performance. This could lead to false positives though since an update that moves a tuple from one component to another doesn't actually generate conflicts.

In Example 12, we've discussed the new incarnation of a tuple with the same primary key might cause confusion. In our context, let's say a transaction T1 reads the last visible version of an item and the tuple is deleted and re-inserted later by transactions T2 and T3 respectively; if the system can't tell the difference between these two incarnations, it might identify it as a RW conflict between T1 and T3 instead of T1 and T2 and could cause unnecessary abortion of transaction T1 or T3 since the last visible version and the newly inserted version are consecutive in the versioning system(because a deletion generates the 'dead' version logically, but not necessarily physically). The solution we've suggested in Example 12 was to attach timestamps of creation to different incarnations. For Snapshot Isolation we've built our serializability implementation upon, we have another approach: create and place a 'dead' version between these two versions so that the two version sets are isolated. This works for predicate RW conflicts too. Notice this may not be appropriate for a system that only stores one version in the database like NDB Cluster. Besides breaking the semantics, a garbage collection system that purges these 'dead' versions is also needed. But in MVCC system like Snapshot Isolation, such a garbage collector already exists. We just need to add the 'dead' versions to the set to be purged.

Now let's look at the following optimized modified write(T, x, xNew) for item RW conflicts:

	  modified write(T, x, xNew) for item RW conflicts:

          get lock(key=x, locker=T, mode=WRITE)

          if there is a SIREAD lock(rl) on x
              with rl.owner is running or commit(rl.owner) >= begin(T):
                  if rl.owner is committed
                      and rl.owner.inConflict:
                          abort(T)
                          return UNSAFE_ERROR
                  set rl.owner.outConflict = true
                  set T.inConflict = true

          existing SI code for write(T, x, xNew)
          # do not get write lock again

          release SIREAD lock(rl) on x if present
	

Remark: The condition 'commit(rl.owner) > begin(T)' has been altered to 'commit(rl.owner) >= begin(T)' to adapt to the new definition of 'concurrent transaction' with a distributed clock as discussed in subsection 5.1.

From the discussion of optimized modified read(T, x) for item RW conflicts case, we know that this write must be the first one after x's read if a SIREAD lock for x exists. So they constitute an item RW conflict. After the write is completed, we may release the SIREAD lock(rl) on x since we don't need that for a later write according to type D of the Serializability Theorem.

In the following optimized modified write(T, x, xNew) for predicate RW conflicts, we can't release the SIREAD lock after the write is completed since other writes in the range might need it later. So the pseudo-code below looks similar as the original:

	  modified write(T, x, xNew) for predicate RW conflicts:

          get lock(key=x, locker=T, mode=WRITE)

          if there is a SIREAD lock(rl) on a range that writing of x is the first match change,
              with rl.owner is running or commit(rl.owner) >= begin(T):
                  if rl.owner is committed
                      and rl.owner.inConflict:
                          abort(T)
                          return UNSAFE_ERROR
                  set rl.owner.outConflict = true
                  set T.inConflict = true

          existing SI code for write(T, x, xNew)
          # do not get write lock again
	

Combining both the item and predicate cases, we got the following algorithm for the optimized modified write(T, x, xNew) case.

	  modified write(T, x, xNew):

          get lock(key=x, locker=T, mode=WRITE)

          if there is a SIREAD lock(rl) on x 
              or there is a SIREAD lock(rl) on a range that writing of x is the first match change,
              with rl.owner is running
              or commit(rl.owner) >= begin(T):
                  if rl.owner is committed
                      and rl.owner.inConflict:
                          abort(T)
                          return UNSAFE_ERROR
                  set rl.owner.outConflict = true
                  set T.inConflict = true

          existing SI code for write(T, x, xNew)
          # do not get write lock again

          if present, release SIREAD lock(rl) on x, 
          but not those on a range that writing of x conflicts with
	

The pseudo-code for modified begin(T) and modified commit(T) remains the same in this optimize algorithm. The arguments for the correctness of this optimized algorithm are the same for the modified algorithm in [Ca 08].

Notice any transaction record must be kept until all the active and future transactions in the system have greater starting timestamps than its owning transaction's commit timestamp so that they can't be concurrent with the transaction in concern. In a distributed system, the garbage collector need to query all the necessary nodes to come to the conclusion that all active transactions satisfy this condition before a transaction record can be purged. This happens asynchronously and should not impose a performance hit.

However, this only guarantees all active transactions are fine. What about the future ones? A distributed clock usually doesn't service timestamps in a monotonic manner. So could a future transaction be assigned a starting timestamp older than the commit timestamp of the transaction record's owning transaction? For instance, in the contrived example when we discussed Lamport's logical clock, a node being inactive for a long time could generate such a timestamp. The solution is again to timestamp the garbage collection messages or the heartbeat messages to find out the smallest possible value for the maximum timestamp each node has issued. If all these values are greater than the commit timestamp in concern, then we are fine since this clock algorithm is monotonic at each node. A similar strategy can be applied to other distributed clocks listed earlier, except the 'happened before' distributed clock. For the 'happened before' distributed clock, if the Monotonicity Condition is satisfied, it will generate timestamps monotonically at each node too. But the discussion in Appendix B indicates that it is not always true. If you are going to use a distributed clock other than those discussed in this article, this issue must be addressed. For the rest of this section, we'll assume this technique is in place and we don't need to worry about these future transactions once all the active ones have been checked up.

It turns out this purging rule also applies to SIREAD locks that haven't been released in the primary algorithm too. Let's see why from the following

	                                                                                                       ## 
	  

Garbage collection of field and 'decision set' versions versions is similar, but we express it in a different way: as long as the distributed clock has advanced to after the timestamp ts of the next version of a concerned version and no transaction in the system has a start timestamp <= ts. For a distributed clock that gives out timestamps monotonically at each node, the first half of this condition implies each node's maximum generated timestamp is greater than ts.

So we use a flag called SIREAD lock to capture the 'Read' part of a RW conflict. Can we do that with the 'Write' part too and replace the write lock with another flag? After all, one attractive feature of an optimistic concurrency control mechanism is that we don't have to build a locking system. It turns out we can. In the Snapshot Isolation we've defined in the last section, updates to both the same field or the same 'decision set' are required NOT to be concurrent and this renders the write locks unnecessary. A flag to replace a write lock must also be kept until all the active and future transactions in the system have greater starting timestamps than its owning transaction's commit timestamp. An implementation of this flag system will be given in the next subsection.

With the optimization from type D of the Serializability Theorem, there will be less false positives when it comes to aborting transactions to prevent conflict loops from forming. But the fact that a conflict loop implies existence of 'dangerous structure', but not the other way around still will give rise to some of them.

Various ways to optimize the primary algorithm are discussed in Cahill's paper ([Ca 08]). I include them here for your reference. In Cahill's primary algorithm, they always abort the pivot(T2 in Theorem 2.1 of [Fe 05]) if either end of a RW conflict can be aborted. One may choose a different strategy like aborting the younger transaction so that long running ones are more likely to survive. Also for simplicity, in the primary algorithm, the abortion decision is postponed till the transaction's commit. But one can do that once a 'dangerous structure' is detected. Likewise, conflicts are not recorded(with T.inConflict and T.outConflict) against transactions that have already been aborted or that will due to both flags being set.

At last, as a solute to Cahill's fine work, I will include one of my own here: detect 'dangerous structure' that is not in a cycle and avoid aborting those three transactions when the primary algorithm executes if a pre-analysis is possible(like in the case where the transactions in an application are known).

5.5 A flag system

We'll provide a sketch of an implementation for a flag system to capture the RW conflicts as promised in the last subsection with a design that starts with the Observer design pattern as described in the 'gang of four' book([Ga 94]). The diagrams for class inheritance are depicted as follows:

For the item RW conflict case, a ConcreteSubject object representing a field version is instantiated whenever a read of that version or the write of next version happens for the first time. Each such read is represented as a ConcreteObserver, which is stored in the list observers inherited from the Subject class. The transaction that performs this read and the time when it happens are recorded in the two variables in ConcreteObserver. The Notify() method is implemented as follows:

      for all o in observers {
        o.Update()
        Detach(o)
      }
	

The subjectState field is used to represent the write of next version. It's initialized to be NULL and updated when that write commits. For simplicity reason, its type is also assumed to be ConcreteObserver in this sketch. The two firlds: transaction and timestamp in ConcreteObserver become the transaction that performs this write and the time when it happens, of course. You may add a new type for it if you prefer, but the diagrams need to be altered accordingly. It is updated in the SetState(Observer o) method when the write of next version happens as follows:

      if subjectState == NULL {
        subjectState = o
      }
      …
      Notify()
	

The Attach(Observer o) method is implemented as follows:

      if subjectState == NULL {
        observers.Add(o)
      } else {
        o.Update()
      }
	

The Update() method is implemented as usual:

      observerState = subject.getState()
	

Here the subject field is inherited from the Observer class and getState() simply returns subjectState. Inside method Update(), this.transaction's outConflict and observerState.transaction's inConflict are both updated to be true. In a typical operation scenario of the flag system, a bunch of reads of a field version come in first and each register itself in the observers list, then a write of next version happens and clears the list; after that, every read of that field version is recognized as causing a RW conflict and doesn't need to be added to the observers list any more.

In a typical implementation, we may arrange the access to subjectState to be atomic, while that to the list observers is protected by a semaphore. But then consider the following scenario: transaction T1 reads a field version(Attach(Observer o) is called in that process) and discovers that subjectState is NULL and tries to add an Observer to the list; soon after that, transaction T2 writes next version of that field(SetState(Observer o) is called in that process) and tries to notify every Observer in the observers list; if for some reason(the thread that executes T1 got context-switch out after subjectState is accessed, for example), T2 acquires the semaphore of the observers list before T1 does and clears the observers list before the addition of the Observer for T1 to it. This violates the semantics of the typical operation scenario described in the last paragraph by adding more Observers to the list after it is cleared. To circumvent this, we may require SetState and Attach to request an exclusive semaphore x before they access subjectState and observers and to relinquish it after that. The following are the resulted code for them:

SetState:

      sem_get(x)
      if subjectState == NULL {
        subjectState = o
      }
      …
      Notify()
      sem_give(x)
	

Attach:

      sem_get(x)
      if subjectState == NULL {
        observers.Add(o)
      } else {
        o.Update()
      }
      sem_give(x)
	

Locality is important. We now have a ConcreteSubject object for each field version of a field. To take advantage of temporal locality, we may place all the field versions of that field into one single ConcreteSubject object. This way we will have as many observers and subjectState pairs as the number of versions of that field in it. We may even use one single exclusive semaphore to control access to all these pairs if we want to reduce memory footprint, at the cost of lower concurrency.

Usually all fields in a logical unit are accessed at the same time. We may take advantage of spatial locality if we place field versions of all these fields in one ConcreteSubject object. We may also consider using one single exclusive semaphore for all of them if not too much performance hit is imposed. If nearby rows in an index are often accessed together, we may go the extra mile by placing the relevant field versions in these rows into one ConcreteSubject object. For example, in TiDB, placing the relevant fields in the rows of a Region into a ConcreteSubject object may not be a bad idea(if a Region is too large, try to divide it into smaller blocks); in a system like NDB Cluster, however, this may not necessarily be beneficial since this part of spatial locality is at least partially destroyed by hashing.

I truly hope that future serializability implementations can make all these – what field versions to place into a ConcreteSubject object and how to use semaphores to control access – configurable to developers. After all, who knows whether an application can take advantage of locality better than its developer? Or even better if the implementation could provide a logical unit version counterpart to field version so that a developer can choose it if he/she is certain that the application accesses data strictly through logical units. This approach can also reduce memory footprint by reducing the number of Observer objects in the system.

Even so, there still could be too many Observer objects in the system that might lead to resource exhaustion. PostgreSQL handles this with the following strategy: if too many tuples in a page have acquired a tuple SIREAD lock, aggregate them into one page SIREAD lock; if too many page SIREAD locks are in the system, aggregate them into one relation SIREAD lock. The details can be found here in PostgreSQL's wiki(at the end of “1.5 PostgreSQL Implementation”). Future flag system implementation could provide a similar mechanism, with an additional field level in the picture of course. Such an approach will nullify our effort of resolving conflicts down to field level and lead to false positives that would not have surfaced, hence should only be considered a measure of last resort. Before that, the system should provide enough amount of memory for this purpose. Luckily, today's high-end servers can already host memory up to a few Tbs. There should be more than enough memory for us, especially when the data in the database are stored on disk. It is a good idea to make the maximum amount of memory for the flag system to allocate Observer objects to be configurable so that database application developers can make their own decision since once again, who knows the application better than them? Designing the Observer objects as compact as possible also helps.

In the predicate RW conflict case, when a predicate read of or a write to range R happens, a ConcreteSubject class is instantiated. The predicate reads are registered in the observers list as in the item RW conflict case, while subjectState becomes a list of objects representing writes in this range. We need a new concrete observer for this kind of object this time. So we have the following diagrams:

The methods to access subjectState are different too.

The implementation of Notify(Observer oForWrite) is as follows:

      for all oForRead in observers {
        oForRead.Update(oForWrite)
      }
	

Here oForRead and oForWrite are of types ConcreteObserver and ConcreteObserverForWrite respectively. This is similar for the implementation of NotifyWrite(Observer oForRead):

      for all oForWrite in subjectState {
        oForWrite.Update(oForRead)
      }
	

The implementations of the Update(Observer) methods in the two concrete observer classes are symmetric. In ConcreteObserver, the Observer oForWrite passed in is a ConcreteObserverForWrite object that represents a write; the Update(Observer) method examines whether that write is the first match change for the read represented by its calling Observer oForRead. On the other hand, in ConcreteObserverForWrite, the Observer oForRead passed in is a ConcreteObserver object that represents a read; the Update(Observer) method examines whether the write represented by its calling Obseever oForWrite is the first match change for it. If it's affirmative, outConflict of the reading transaction and inConflict of the writing transaction are both set.

Both observers and subjectState are guarded with a shared semaphore for access. Let's called these two semaphores s.o and s.s respectively. Then the implementation of method AttachWrite(Observer oForWrite) is as follows:

      sem_get(s.s, exclusive_mode)
      subjectState.Add(oForWrite)
      sem_give(s.s)
      sem_get(s.o, shared_mode)
      Notify(oForWrite)
      sem_give(s.o)
	

Symmetrically, the implementation of method Attach(Observer oForRead) is as follows:

      sem_get(s.o, exclusive_mode)
      observers.Add(oForRead)
      sem_give(s.o)
      sem_get(s.s, shared_mode)
      NotifyWrite(oForRead)
      sem_give(s.s)
	

If we view observers and subjectState as two data fields and the access of them in methods AttachWrite and Attach as two transactions, it is possible to find a conflict loop between them when they are executed concurrently and hence it is NOT serializable by type D of the Serializability Theorem. However, the semantics that each and every predicate RW conflict is captured remains correct since each possible predicate RW conflict is evaluated by either AttachWrite or Attach. That said, there is a chance that a predicate RW conflict is evaluated twice in this process. But that doesn't change the correctness of the algorithm since the updates to outConflict and inConflict are idempotent.

In some implementations, ranges could be merged or split. For example, in TiDB, two Regions can be merged if there are two few tuples in them and a Region can be split if it gets bigger than its default size of 256 MB. The algorithm must be able to cope with this. For the merging case, two relevant ConcreteSubject objects are merged into one, so are the observers lists and the subjectState lists in them. For the splitting case, the observers list is copied to both newly generated ConcreteSubject objects, while the subjectState list is split into two so that each only contains writes for its sub-range.

So far, this works fine for a predefined range R that is associated with a known object like a Region in TiDB, a page in Cahill's prototype of Serializable Snapshot Isolation deployed in Berkeley DB or a tuple in the added table for predicate RW conflicts in the serializability implementation I've developed for NDB Cluster. But in other implementations when this association is missing, there is an extra issue. For example, in the CockroachDB case, the actual range(or its components) inside a where clause is recorded in a node-local cache to handle predicate RW conflicts. Say, in a statement “… where key>=1 and key<5 or key>100 and key<200”, [1, 5) and (100, 200) are inserted into its target node-local cache respectively. If we use a range R for [1, 5) in the primary algorithm and an insert into it “happened before” the predicate read, but they still constitute a predicate RW conflict(for example, the transaction with the predicate read starts with a earlier timestamp and it's a long-running one, the predicate read happens very late in that transaction); then when the insert comes in, we don't know where to put it since the range R for [1, 5) hasn't been determined yet. To get around this, we need to provide an auxiliary data structure like a list to hold such writes until an appropriate predicate read shows up, or otherwise it's garbage-collected.

So it's time to talk more about garbage collection. Some objects in observers and all objects in subjectState are never deleted in the primary algorithm. They need to be garbage-collected when they are not needed any more. For those inside observers, they are just SIREAD locks and the rule for garbage-collecting them has been discussed. For an object inside subjectState, the rule for garbage collection is also the same: it can be purged as long as no active or future transactio in the system has a start timestamp <= timestamp of it; this way, all the possibly conflicting SIREAD locks have been processed against it before garbage collection. The Detach and DetachWrite methods are used in this garbage collection process. This rule also applies to the writes in the auxiliary data structure.

We also need a buffer pool to store pre-located and initialized ConcreteObserver and ConcreteSubject objects for performance considerations, especially in a system that doesn't support slab allocation.

Notice in this flag system, every read is accompanied by a write(of an Observer object), at least. In contrast, in a database with pessimistic technology, only a locking read is accompanied by a write(the lock request in the locking system). This may impact performance, of course. Future implementation should provide tools for pre-analysis so that unnecessary writes to the flag system(for example, if a field is only read, but never written or the other way around) are not possible. I understand Serializable Snapshot Isolation poses as a serializability implementation that can handle workload that is unknown at the time of development. But sometimes performance trumps and important workloads usually don't allow arbitrary transactions to be fed into the system for safety reasons. Providing these pre-analysis tools certainly helps with this latter situation. For example, if these tools are good enough, we don't even need to activate the flag system for TPCC since it will be detected that it doesn't contain any 'dangerous structure'.

So far the flag system works if all transactions commit successfully. But what if a transaction got aborted prematurely, or the system crashes? For example, what happens if a transaction T is aborted after a SIREAD lock is added to the observers list and the subjectState list is being notified? We'll try to answer the following two key questions should T be aborted prematurely, or the system crash:

1. Does the flag system still guarantee correctness by destructing each and every 'dangerous structure'?
2. Can the flag system maintain a correct state when this happens?

We first look at the case when a transaction got aborted prematurely. In the example we've just described, we may just remove it(with the Detach method) when its containing transaction is rolled back. Some objects in the subjectState list may be deleted(with the DetachWrite method) when their containing transactions are rolled back because their associated inConflict fields are set. Although these transactions are aborted for some transaction that never commits, it only generates false positives for presence of 'dangerous structure's. So the answer to the first question is affirmative. The flag system remains in a correct state after T is rolled back since all the objects for the relevant transactions that can't commit are removed.

In the case where the system crashes, for the first question, if the recovery mechanism is like Write-Ahead-Logging so that no committed transaction is lost when the system crashes, all we need to worry about are those transactions that haven't committed when disaster strikes. Some of them may have caused other transactions to abort, but that is it. In particular, they don't roll back committed ones and hence doesn't affect the surviving ones. So the answer to the first question is again affirmative.

If, on the other hand, the recovery mechanism is like neighborhood-Write-Ahead-Logging as in NDB Cluster so that some committed transactions are lost in the crash, the answer to the first question is still affirmative. The reason is simple: if all committed transactions can't generate any 'dangerous structure', how could some of them?

The second question doesn't apply in either case if the system crashes since upon system restart, the flag system-which exists in memory only-will be recreated from scratch. We've used an example for demonstration here. But actually it works for all other situations.

Notice this flag system implementation doesn't require a join to be re-written as equivalent statements as in the serializability implementation I've developed for NDB Cluster. That is because conflict resolution happens when a predicate read is on-going, which is after an execution plan is chosen.

5.6 False Positives

Some steps in this algorithm could generate false positives for conflict loops. We'll identify them and try to provide alleviation if possible.

1. Having a 'dangerous structure' doesn't mean it is in a conflict loop. An implementation may provide tools for pre-analysis of conflicts so that the obvious false positives can be discovered beforehand. And it should provide means to annotate the relevant conflicts such that the primary algorithm can skip them.
2. Aggregation of finer granularity SIREAD locks into one coarser granularity SIREAD lock could also lead to false positives.
3. Range R chosen to cover a predicate read is too large and this could lead to false positives. Make sure this cover is as 'tight' as possible is an alleviation. For example, in TiDB, if a Region is too large, we may use a sub-region of it as R if that is supported.
4. If range R is scattered into different nodes, an update that moves a tuple from one component to another could lead to a false positive. To eliminate this type of false positives, we need to globally monitor R. This may induce a performance hit.
5. The last type of false positives are those caused by a transaction that never commits as we've just demonstrated in last subsection. One way to eliminate some of them is to set the inConflict and outConflict fields in the primary algorithm right before T commits(delay calling of Update() methods until then), so that those transactions aborted early are not causing any false positive.

This last technique of reducing false positives is relevant to the answer of the following question: can we move the flags up and down in a transaction? In the serializability implementation I've developed for NDB Cluster, we can always move the locks for prevention measures up the transaction, and sometimes move them down. What is the situation in a flag system? Are there any implications of doing that?

It turns out we may move them up or down freely. That is because altering their locations in a transaction only changes the time they are inserted into the observers and subjectState lists, not the outcome of the methods those actions triggered. We've just seen the advantage of moving these flags downward: potentially less other transactions are rolled back because they may have already committed when the flags are set. The disadvantage, of course, is that this transaction itself might get aborted when a 'dangerous structure' exists since the counterpart has already committed and can't be rolled back any more. In this case, some less work could be saved should this transaction set those flags and got rolled back early.

5.7 Summary and discussion

Let's summarize what we've accomplished in generalizing Serializable Snapshot Isolation to a distributed system:

1. A field-based Snapshot Isolation for a distributed system is defined in Section 4.
2. On top of it, type D of the Serializability Theorem is proved in Section 4. It can be applied to a distributed system as other types.
3. Two alternatives are provided to actually implement the underlying field-based database system in this section. A distributed clock satisfying the Clock Condition can be chosen to implement a field-based Snapshot Isolation on top of it. 'Decision set' versions are captured as demonstrated in Section 4.
4. The Clock Condition guarantees that arguments in [Fe 05] can go through in a distributed system. This time, type D of the Serializability Theorem is used to conclude serializability instead.
5. A flag system is sketched to realize the book-keeping algorithm in [Ca 08].
6. Different choices of distributed clocks and underlying database implementations give rise to different systems. And the flag system is also an alternative to the hybrid system employed in PostgreSQL. And the generalization is hence generic.

So now we have a Serializable Snapshot Isolation implementation without using any long locks. Can we use long locks to replace the flags in the flag system? If we only replace the read flag or the write flag with a long lock, then we need a hybrid system that involves both flags and long locks. That is exactly what PostgreSQL does and the details are described in the wiki here. So can we replace both with long locks? We can, but it turns out there is a catch.

From the implementation of the flag system, we know that a SIREAD lock can't be purged after its containing transaction T1 commits since another transaction T2 started before T1's commit may perform a write after T1's commit that conflicts this SIREAD lock. If T2 is long-running and we use a long lock as a SIREAD lock, this SIREAD lock will block writes conflicting with it for a very long time. This is certainly undesirable since Snapshot Isolation based technology always poses as one that writes don't need to wait for reads on the same data item. PostgreSQL‘s team knows about this and it is mentioned in its wiki(in the section “4. Innovations”). I don't know if Cahill's team knows about it. It is not mentioned in [Ca 08].

Although this section is based on type D of the Serializability Theorem, you may also apply the method developed in it to a type B of the Serializability Theorem as long as the field versions are derived. Specifically, if you already have a distributed Snapshot Isolation implementation based on tuple versions, with one of the distributed clocks discussed in this section, all you have to do is to implement the primary algorithm and the flag system on top of it and achieve serializability. That's exactly what we could've done with alternative one. Concurrency control is not as fine-grained as with type D of the Serializability Theorem, but the alternation to your code is minimized.

In general, there could be other specific issues you need to work out when you try to develop a Serializable Snapshot Isolation implementation in a distributed database system. For example, in the second-tier serializability implementation I've developed for NDB Cluster, I also work on a subsection called 'Durability of consistency' so that more guarantees are provided in the case a NDB Cluster suffers a system-wide crash. Such work would require a lot of details about the system to fit in so that it could go through. If you are trying to work out an issue in the process and encounter problems, please don't hesitate to contact me if you think I may be able to help.

From what we've discussed in this section, depending on the specific implementation, it may or may not require long locks other than semaphores for concurrent writes to the same field. But in the case it may, we need a field level capable locking system for finest concurrency control.

7.1 Questions to be answered

Next let's explore a bit the questions haven't been answered in this article. I hope answers to them will help shape the future of the field of consistency.

When multiple transactions are trying to update the same item, it creates a race condition. There are currently two approaches to resolve this race condition: let them race or let them wait. The 'let them race' approach aborts all the transactions that lose the race, with only one winner. In a hot spot problem scenario where a lot of transactions are trying to update the same field simultaneously, the 'let them race' approach will lead to cascading abortion, which is as bad as the thrashing behavior demonstrated by 2PL. Hence implementation like Snapshot Isolation 'optimistically' assumes a race condition would not happen. This is the so-called 'optimistic technology' which underlines consistency model in important commercial databases like Oracle, Microsoft SQL server and PostgreSQL. The 'let them wait' approach assumes a race condition could happen and let the contenders wait for their turn. This is the so-called 'pessimistic technology'. MySQL InnoDB and NDB Cluster both implement it.

As a second-tier developer, I've never been a big fan of this 'optimistic technology ' approach. That is because if one assumes the role of a database application developer long enough, she/he will encounter the hot spot problem and need to handle it. So if one starts with 'optimistic technology', she/he will eventually have to take a look at 'pessimistic technology'. So why don't we start with the latter in the first place?

	               T1                                                              T2

          start transaction;

          select value into :value from t where id1=1;

                                                                              start transaction;

                                                                              update t set value=25 where id1=1;

                                                                              commit;

          update t set value=:value^2 where id1=1;

          commit;                                                                                                 
	    

	                                                                                                       ## 
	    

7.2 Hot spot problem

While we are at the topic, let's explore the hot spot problem a little bit. We start with an example showing how bad it could be if 'optimistic technology' encountered the hot spot problem.

        update t set ticket=ticket-1 where ...;.                                                                                                 
	  

        e = Real work done by the system / Total work done by the system.                                                                                                   
	  

	                                                                                            (To be continued ...)  
	  

In an ideal world, in a database system where 'pessimistic technology' is implemented, each transaction in a race condition will just wait in a lock queue quietly for its turn. This way in a thrashing system, although the wait for a lock is so long that it appears that the system is stalked, it's actually progressive and the transaction in wait will eventually be served. The reality, of course, is way more complex than the ideal case.

First, the update transactions flooding in might saturate other resources of system and practically render a DOS attack. For instance, if the queue of lock wait is too long, it might consume all the system memory and direct the system into an erroneous state.

Second, 'pessimistic technology' like NDB Cluster usually uses time-outs to detect deadlocks. This mechanism will time-out those statements that have been waiting too long for a lock. If we employ a re-try mechanism to restart these dead-lock victims, this effectively re-generates the cascading abortion scenario discussed in Example 29.

So to employ 'pessimistic technology' to deal with the hot spot problem, we must use a different deadlock detection/elimination strategy.

In the literature, besides using time-out to eliminate potential deadlocks, people have also explored lock dependence tracing to detect deadlock. In this approach, lock dependence between transactions are surveyed regularly to find out if a dependence loop forms and one of the transactions in the loop will be aborted if the result is affirmative. This algorithm usually implies a Depth-First-Search(DFS) in an inner loop. For a dense graph, DFS is an O(n^2) algorithm. So this lock dependence tracing mechanism may not be very practical. I am currently not aware of any commercial database system employing it.

There is another deadlock detection/elimination strategy in the literature. We may impose a total order on the locks acquired by transactions in an application. For example, if that application consists of five tables, we impose a total order on those tables; and when we try to acquire locks in a specific table, we also follow a total order – the order of the primary key could be one such choice; and if we implement field level capable locking system in the future, we may also need to impose a total order on the fields in a tuple for the same purpose. Notice that if we were to implement it so that type D of the Serializability Theorem can be applied, the 'decision set' locks must also be part of this order. Putting them before the field locks of the same row and after those of the previous row should be a good choice. This way, a deadlock would never happen and so this is a deadlock elimination strategy.

In this strategy, the order of acquiring locks in a transaction may not be the same of the order of executing the statements requiring those locks. This may imply we need to use a strategy to execute the transaction like that in the original 2PL design: we acquire locks in the first phase, we then execute the statements, and we release those locks in the second phase – it is also called conservative 2PL in the literature. In such a design, we need to access the page containing a data item twice, the first time to lock it(we at least need to see if it is there before we lock it) and the second time when we execute the statements. A disk-based database may suffer a performance hit since it might need to bring in the page twice to the memory pool in some cases. For example, after it is brought in for the first time, it becomes a victim of the eviction algorithm of the memory pool and it must be brought in a second time. For a memory-based system like NDB Cluster, this performance hit should be marginal.

Care must be taken when we implement such a design since besides the statements in a transaction, the so called DML statements, there are also DDL statements in a database system. If we only impose the total order for DML statements locks and not for DDL statements, a deadlock situation could still result; and because we don't have a deadlock detection mechanism in this scenario, the system will stalk when it happens.

So this is one possible future, a future where the hot spot problem can be handled by 'pessimistic technology'. But right now, if we are asked the question if we could handle the hot spot problem while providing serializability, the answer is probably negative. That is because Serializable Snapshot Isolation is born with the assumption that it doesn't have to deal with it, 2PL and Google's Spanner will demonstrate thrashing behavior and the method I've developed for NDB Cluster will time-out transactions that should have been waiting quietly for their turn. So it seems we don't have a very good solution with only third-tier technology(only if you agree that the method I've developed for NDB Cluster is mainly third-tier technology).

Here enters the sophisticated second-tier database application developer.

In my opinion, the hot spot problem is really a resource problem, in which people are competing to gain control of precious resources, like the super bowl tickets in Example 29. So the real solution is to increase resource providing. Unfortunately that is not always possible, like the stadium for the super bowl match has only a limited number of seats. When it is impossible, it is the database application developer who will catch the ball. Since it is not a computer science problem, we can't guarantee solving it completely by applying computer science methods and sometimes providing an alleviation is already the optimal solution. One approach to this end is to distribute the load to a few sites so that at each site the race become less intensive.

        update t set ticket1=ticket1-1;

        update t set ticket2=ticket2-1;

        update t set ticket3=ticket3-1;

        update t set ticket4=ticket4-1;

        update t set ticket5=ticket5-1;.                                                                                               
	  

	                                                                                                       ## 
	  

However, this simple 'distributing the load to a few sites' approach breaks down as soon as the situation becomes slightly more complex. Suppose in this Example 29, a restaurant boss in the stadium wants to conduct a promotion by providing 100 free meals for the first 100 tickets sold for the 5000 VIP seats. So the following statement is added to the transaction:

      update t1 set free_meal=free_meal-1 where ...;.                                                                                                   
	

Here free_meal starts with an initial value 100. Of course, you may distribute the 'free_meal' field to the five sites like the 'ticket' field case so that the initial value at each site is 20. But then it is possible that at some sites, not all 20 free meals may be given out. For example, a site was initially set up NOT to sell VIP tickets, but that was changed right before the sell and the message wasn't communicated to the potential VIP seat buyers well enough. Then the total number of free meals given out could be less than 100. Now what if the restaurant boss is a very stringent one and considers the promotion unsuccessful and refuses to pay for it? In general, if we have some sort of global requirement for the data item in a race condition, it is hard to distribute it to different sites.(In this case, the global requirement for 'free_meal' is that it must reach value 0 at the end and we don't have a similar requirement for the 'ticket' field)

So if the 'distributing the load to a few sites' approach doesn't work, what do we do? The sophisticated second-tier database application developer might again got an answer: we may implement a control mechanism, a queue for example, at the second tier so that only a bounded number of transactions causing the race will be fed to the third tier per unit time. When the queue is saturated, we may print out messages like 'The server is busy, please try again later.' and reject the request. Those transactions that don't cause a race will pass through without blocking. This requires a detailed analysis, often a pressure test on your application to find out what would stress your system and what would not to design this control system. In general, handling the hot spot problem successfully requires collaboration between both the second tier and the third tier. Since this is not the primary subject of this article, I won't expand in this direction any more. You are very much welcome to discuss it and share your experience with me though. This is an area I find intriguing.

7.3 Security implications

To round up this section, I'd like to discuss the security implications of consistency a little bit. At the beginning of this article, I mentioned the Flexcoin incident. Now let's see how to mount such an attack in a Read-Committed isolation level like that of NDB Cluster's. According to [Si 14], the transaction of transferring funds between accounts from Flexcoin contains something like the following SQL snippet(We've put back a check on newbalance that is skipped in [Si 14] before the write to it here for our purpose. The semantics will be slightly different from that in [Si 14]:

      mybalance = database.read("account-number")
      newbalance = mybalance – amount
      if newbalance >= 0 {
        database.write("account-number", newbalance)
        dispense_cash(amount)   // or send bitcoins to customer
      }.                                                                                                  
	

Now suppose two transactions are executed by the same customer on the same account concurrently and interleaved like this:

	               T1                                                              T2

      mybalance = database.read("account-number")

      newbalance = mybalance – amount

                                                                        mybalance = database.read("account-number")

                                                                        newbalance = mybalance – amount

      if newbalance >= 0 {

        database.write("account-number", newbalance)

        dispense_cash(amount)

      }

                                                                        if newbalance >= 0 {

                                                                          database.write("account-number", newbalance)

                                                                          dispense_cash(amount)

                                                                        }                                                                                                   
	

A conflict loop consisting of a RW conflict from T2 to T1 and a WW conflict from T1 to T2 on the implicit 'balance' field forms and it is a 'lost update' scenario. For example, assume the original account balance is 2 bitcoins and amount = 1 bitcoin. After both transactions commit, the account balance is 1 bitcoin since one of the deduction is 'lost'. But totally 2 bitcoins are sent out since both transactions are successful.

In the attack that actually happened, the threat actor first created an account with a small amount of bitcoins in it. She/he then magnified this 'lost update' effect by sending thousands of transactions to this account concurrently. When the account balance reaches zero and the transactions stopped sending out bitcoins, an amount much larger than the original small balance was already sent out. The threat actor then created more accounts and repeated this process until Flexcoin's hot wallet was drained.

We may replace the fourth line in the snippet with something like the following update to see what happens, assuming 'balance' to be the field in the database that is updated:

      update t set balance = balance – amount 
         where account = account-number;                                                                                                    
	

Although the original RW and WW conflicts still exist and form a loop, the corresponding history is not a 'lost update' scenario any more since the second update actually sees the first update from the implicit read of balance and acts accordingly. After both transactions commit, the account balance will be 0 if it started with 2 bitcoins. So this modification seems to make the application correct, right?

It turns out it only makes it harder, but NOT impossible to mount an attack. Suppose a third transaction is executed and if we could manage to interleave the statements as such: the three database.reads and test conditions are executed first, then the three updates. All three transactions will commit successfully and the 'balance' field will be left with value -1.

Now let's see how to manage this interleaving and mount an attack in the real world. If the client end provided by the bitcoin exchange is open source and the transfer transaction code is in it, all you have to do is to insert a statement like

      sleep(10000)                                                                                                    
	

after

      newbalance = mybalance – amount                                                                                                     
	

and issue, say, 1000 such transactions to the server. Most of the transactions will line up the exact pattern we need. If the client end is not open source but the transfer transaction is still issued there, the threat actor might still manage to achieve this since the client end resides in an environment controlled by the threat actor. If the transaction is issued from an application server instead, the threat actor need to hack it first and it makes it much harder, although not impossible.

Some database application developers try to tackle consistency related issues with naive approaches. For example, for an important field like 'balance', they may impose rules in the application to make sure it is always non-negative or use a trigger on it to take action as soon as it becomes negative, like shutting down the application. The 'lost update' scenario shows that the first approach doesn't work at all in that case. The second approach is better, but the threat actor can still double her/his bet in the best case in the scenario not involving a 'lost update'.

Therefore I highly recommend you to use the systematic way developed in this article instead to fence off this kind of threat. To that end, all four serializability implementations we've explored certainly will do the job since a conflict loop will not be possible at all. It turns out even Snapshot Isolation is good enough for the example in this sub-section since the update of the 'balance' field will already make any two transactions non-concurrent. Notice that we can't apply the Generalized Serializability Theorems here since all the transactions are Read-Write and all the reads in them affect writes.

I will stay in this field for at least five more years and see if we could solve the serializability problem completely in this spin. You are very much welcome to join me in this effort.

A.1 The theoretical framework

In this article, I built a theoretical framework for the Serilizability Theorems on top of Adya's. The process combines the original framework with new concepts and definitions, which might seem messy for some of the audience. I'd like to clarify things up in this sub-section by restating it for your reference.

A transaction is like a program in a programming language like C. It may have input parameters, statements in a regular programming language like branching statements and loops, deterministic math calculations, non-deterministic math calculations(like calling the Rand() function in MySQL) and finally SQL statements. SQL statements are of two types: those that access the database and those that don't. For those that don't access the database, their behavior depends on the environment, just like regular programming language statements; for those that access the database, their behavior also depends on the database state. What is more, no mater what branches the execution might follow, the SQL statements are executed in a sequential order in a transaction.

Database consists of data objects that can be read or written by transactions. The data objects will be dealt with in this article are tuples, fields and most importantly, 'decision set's. Tuples, fields correspond to rows and fields in a table respectively. When we constrain the predicate in a SQL statement to be one that only involves one table, we define the set of columns in the only table that a predicate P uses to decide whether a tuple is selected for later access to be the 'decision set'. More complex predicates involving more than one table, like joins, sub-queries, unions, are handled separately.

A data object has one or more versions. However, transactions only interact with the database in terms of data objects; the database system maps each operation on a data object to a specific version of that object. A transaction may read versions created by committed, uncommitted, or even aborted transactions; constraints imposed by some isolation levels will prevent certain types of reads, for example, those created by aborted transactions.

When a transaction writes a data object x, it creates a new version of x. A transaction Ti, i>=0, can modify a data object multiple times; its first modification of data object x is denoted by xi:1, the second by xi:2, and so on and they are called internal versions. Version xi denotes the final modification of x performed by Ti before it commits or aborts. A transaction’s last operation, commit or abort, indicates whether its execution was successful or not; there is at most one commit or abort operation for each transaction. The committed state reflects the modifications of committed transactions. When transaction Ti commits, each version xi created by Ti becomes a part of the committed state and we say that Ti installs xi; the system determines the ordering of xi relative to other committed versions of x. If Ti aborts, xi does not become part of the committed state.

Conceptually, the initial committed state comes into existence as a result of running a special initialization transaction, Tinit. Transaction Tinit creates all data objects that will ever exist in the database; at this point, each data object x has an initial version, called the unborn version. When an application transaction creates a data object x (e.g., by inserting a tuple in a relation), we model it as the creation of a visible version for x. Thus, a transaction that loads the database creates the initial visible versions of the objects being inserted. When a transaction Ti deletes a data object x(e.g., by deleting a tuple from some relation), we model it as the creation of a special dead version, i.e., in this case, xi is a dead version. Thus, data object versions can be of three kinds: unborn, visible, and dead.

If a data object x is deleted from the committed database state and inserted later, we consider the two incarnations of x to be distinct objects. When a transaction Ti performs an insert operation, the system selects a unique data object x that has never been selected for insertion before and Ti creates a visible version of x if it commits.

We assume data object versions exist forever in the committed state to simplify the handling of inserts and deletes, i.e., we simply treat inserts/deletes as write (update) operations. An implementation only needs to maintain visible versions of data objects, and a single-version implementation can maintain just one visible version at a time. Furthermore, application transactions in a real system access only visible versions.

A history H over a set of transactions consists of two parts: a partial order of events E that reflects the order of the operations (e.g., read, write, abort, commit) of those transactions, and a version order, <<, which is a total order on committed versions of each data object.

Finding a way to impose such a total order on committed versions of each data object should be easy. For instance, in a system with a field level capable locking system, we may use the order of locking events on a field to introduce such an order; for a system with Snapshot Isolation that is lock free, the complete isolation in time of two transactions updating the same field should give rise to such an order. In the case tuple versions are present, the situation is similar if a total order of tuple versions is used; otherwise an order of field versions is derived from the tuple versions. For 'decision set' versions, they are either derived as in the type B case or defined as in the type D case.

Besides the committed versions, we also need to define an order for the internal versions of a data object inside a transaction in the case it is modified more than once. We've explained earlier before Example 1 that the execution of a transaction is a sequential process, namely, on a statement by statement basis. So assuming each statement doesn't update a data object twice, the order of these versions is obvious, for either field, tuple or 'decision set' versions.

The partial order 'happened before' is defined in Example 6. In general, we may impose a stronger condition on the write events of the same data object by requiring that they happen in a serial order so that it coincides with the 'happened before' partial order. This implies the version order of the committed versions coincides with the 'happened before' partial order too, namely, one version is before another if and only if it 'happened before' another.

A write operation on data object x by transaction Ti is denoted by wi(xi)(or wi(xi:m)); if it is useful to indicate the value v being written into xi, we use the notation, wi(xi, v). When a transaction Tj reads a version of x that was created by Ti, we denote this as rj(xi) (or rj(xi:m)). If it is useful to indicate the value v being read, we use the notation rj(xi, v). In a committed transaction, wi(xi) writes the committed version xi.

Notice the versions in Vset(P) could be either committed or internal ones and they could be of either field, tuple or 'decision set' granularity. Predicate P is then evaluated against version set Vset(P), the tuples with a version in the version set that satisfies the predicate are chosen for later access. This evaluation process is the realization of a predicate read. A predicate read performed by transaction Ti is denoted by ri(P: Vset(P)) and let's call this subset of Vset(P) that satisfies the predicate the matching set. A write can change the matching set in two ways: it changes a non-matching version outside the set into a matching version inside the set(the IN operation) or it changes a matching version inside the set into a non-matching version outside the set(the OUT operation).

Now let's start listing the conditions for the Serializability Theorems.

Predicate reads are also included in E, isolation of them is Adya's major contribution to this framework.

In Condition 2' we didn't claim wi(xi:m) 'happened before' rj(P: Vset(P)). A predicate read could be a lengthy process, for example a full table scan in some cases, and wi(xi:m) could overlap with rj(P: Vset(P)).

In some Snapshot Isolation implementations, Condition 3' could be violated since every read is supposed to read from the snapshot when the transaction starts. We've demonstrated after the proof of the Serializability Theorem that the proof will go through even so.

Here in Conditions 2' & 3', a version of x could refer to a field, a tuple or a 'decision set' one.

From now on, we'll only consider complete histories. Notice that Condition 4 allows us to only consider committed transactions since atomicity guarantees the aborted ones NOT to have any effect on the system. So we'll only consider systems that can provide atomicity from this point on.

The definition above identifies Ti to be a transaction where a change occurs for the matching set of rj(P: Vset(P)). Note that i can be equal to j. Version xi, xh here could represent a field, a tuple or a 'decision set' version respectively.

Version xi, xh in the last definition could represent a field, a tuple or a 'decision set' version respectively.

Again a version like xi in these two definitions represents either a field, a tuple or a 'decision set' version respectively. Notice there is an extra requirement that Tj doesn't generate the version after xk itself in Definition 4' comparing with Definition 4 in Adya's work. If Tj does generate the version after xk itself, the dependence between Tj and Ti is captured with a sequence of WW conflicts on x.

Again a version like xi in these three definitions represents either a field, a tuple or a 'decision set' version respectively.

Here we combine Definitions 3' & 5, 4' & 6 and call them Directly Read-Depends(WR conflict), Directly Anti-Depends(RW conflict) respectively for convenience.

        w1(x1:i) : : : r2(x1:i) : : : (a1 and c2 in any order)

        w1(x1:i) : : : r2(P: x1:i, ...) : : : (a1 and c2 in any order)
                                                                                        
	  

        w1(x1:i) : : : r2(x1:i) : : : w1(x1:j ) : : : c2

        w1(x1:i) : : : r2(P: x1:i; :::) : : : w1(x1:j ) : : : c2
	  

From this definition, it is clear that the order of transactions coincides with the order of write events of the same objects in the 'happened before' partial order, hence coincides with the total order of the object versions.

We assume that when a transaction reports back to the API coordinator its state as committed, all of its modifications are available for reading.

With this assumption, it's easy to see that a later transaction in the execution sequence sees the latest modifications in transactions before it: the time for the modifications of a previous transaction to be available is before the time for the commit message to reach the coordinator, which is before the time to start a later transaction, which is once again before the time the reading happens; and since the transaction reads the latest available modification, it reads the latest modifications in transactions before it.

On the other hand, an earlier transaction in the execution sequence sees no modification of a later transaction. Suppose the opposite was true and an earlier transaction could read a modification of a later transaction. The write operation 'happened before' the read operation for any reasonable implementation such that Condition 2' is satisfied with the 'happened before' partial order. However, the read operation 'happened before' the commit of the earlier transaction, which 'happened before' the start of the later transaction, which again 'happened before' the write operation. This contradicts with the fact that 'happened before' is a strict partial order.

The assumptions for the four types of the Serializability Theorem are as follows.

Type A: All the conflicts are based on tuple versions. It is used to demonstrate that Adya's Serializability Theorem can actually be proved, without resorting to the results in [Be 87].

For this type, the first internal tuple version generated by a write operation need to be paid more attention to. Besides the fields that are actually written in the write operation, there might be fields that are not. Since we are generating a tuple version, we have to fill in a value for each of those. Where do these values come from?

There are three cases. In the first case, all fields in the tuple are modified and we need to do nothing. In the second case, a tuple read 'happened before' the write in the same statement like in 'update...set col=col+1'; in this case, the values naturally come from the tuple read. In the third case, it is a blind write like 'update table grades set grade=”A”'. In this article, we assume the following:

The third case works the same as the second one except that it reads a tuple version implicitly before the write. In other words, there is no real blind write in our framework.

For predicate-based conflicts, Definitions 1 & 2 remain the same as in Adya's paper. The digression starts with the added Definition for the 'conflict with' concept. The item-based conflicts remain the same as in Adya's paper, of course.

Type B: Updates in transactions still generate tuple versions, both internal and committed ones. Field versions are derived from these tuple versions as usual. All the regular item-based conflicts are interpreted at field level, namely, based on the derived field versions.

Like the derived field versions, we also derive 'decision set' versions from these tuple versions. Namely, a derived 'decision set' version is generated whenever fields in the 'decision set' are modified in a tuple write and these derived 'decision set' versions are ordered the same way as their associated tuple versions.

For predicate reads, we use derived 'decision set' versions to capture the conflicts. This is true for Definitions 1, 2, 3' and 4' and the added Definition for the 'conflict with' concept.

There is a new kind of conflict in this type of the Serializability Theorem though. They are the WR, RW and WW conflicts between reads and writes of the committed derived 'decision set' versions, just defined like those for tuple or field versions. It is easy to see that the serial order of write events on the 'decision set' coincides with the 'happened before' partial order since that of the tuple write events do. After the introduction of this new set of conflicts, we also include them into the item-based conflicts, as opposed to the predicate-based ones. In particular, Definitions 5, 6 & 7 for type B include committed 'decision set' versions now, not just committed field versions.

As in the tuple write case in type A, we may require an implicit 'decision set' version to be read before a 'decision set' version write if not all the fields in the 'decision set' are written. That is because in a system that still generates tuple versions, a 'decision set' version write where not all the fields in the 'decision set' are written implies a tuple version write where not all the fields in it are written. Hence we may require that a tuple read must precede such a tuple write as in the type A case and the derived 'decision set' read will serve our purpose.

For possible item reads following a predicate read, we will use the 'derived' field versions from the same tuple version associated with the matching 'decision set' version.

Type C: This type of the Serializability Theorem still requires the tuple concept which corresponds to the row concept in a table, but it doesn't require the concept of tuple versions. It does require the existence of field versions and a total order generated by serial write operations on that field. All conflicts are interpreted at the field level and this implies that the 'decision set' of a predicate read only consists of one field. It also requires a way to identify which field version to read after a predicate read if item reads follow.

Type C of the Serializability Theorem is designed for the future, for example, when a field level capable locking system is available. Notice that the item-based conflicts between reads and writes of 'decision set' versions are just those between field versions.

In type C, the requirement of the way to specify what field versions to read after a predicate read is matched is implicit: as long as it makes a serial history satisfy its key property, namely, when a transaction is executed, it always reads the latest committed version of data object that is available.

Type D: Type D of the Serializability Theorem assumes a field version system as in Type C of the Serializability Theorem.

On top of it, we need a systematic way to define a 'decision set' version system so that the following requirements are met for type D of the Serializability Theorem.

1. A way to define the system.
2. A way to read the 'decision set' versions in a predicate read without mistakenly using an arbitrary set of field versions in the 'decision set' instead, so that predicate-based conflicts can be defined.
3. A way to read the 'decision set' versions for the initial internal version without mistakenly using an arbitrary set of field versions in the 'decision set' instead, so that successive internal versions and eventually the committed 'decision set' version can be defined.
4. And also a way to introduce item-based conflicts between reads and writes of 'decision set' versions so that we can reason about conflicts with them.

Of course, it also requires a way to identify which field versions to read after a predicate read as in type C if item reads follow.

So this is the framework we've developed in this article. On top of this, we may present the Serializability Theorems and the Ramification Theorems, which I will not repeat here. If you wonder why we make all these design decisions, juicy details and discussions are readily available in the article.

Next, let's demonstrate a weird Read-Committed isolation level to which we can't apply these theorems in the following example.

	                                                                                                       ## 
	  

In type C and type D of the Serializability Theorem, we claimed the requirement for the way to specify what field versions to read after a row is matched in a predicate read is implicit: as long as it makes a serial history satisfy its key property, namely, when a transaction is executed, it always reads the latest committed version of data object that is available. From this example, we see why it is necessary; from the proof of type C and type D of the Serializability Theorem, we see why it is sufficient.

D.1 Google's Spanner

This analysis is based on [Co 13]. Google's Spanner builds on top of a couple of Google's software artifacts like Colossus and Tablet. Upon those, a Paxos system is implemented to achieve eventual consistency. The leader of each Paxos group participates in a variant of Two Phase Locking(2PL) for the Read-Write transactions, while other replicas of that Paxos group provide Snapshot Isolation reads to Read-Only transactions.

In this process, a distributed timing system is necessary to provide timestamp service for Snapshot Isolation reads and other purposes. A variant of Marzullo's algorithm for distributed time described in [Ma 83] is used to synchronize clocks on different time servers, which have been set to initial values by polling a list of accurate atomic clocks and GPS clocks with the algorithm described in [Mi 81]. This time service is an interesting one since it returns an interval containing the correct time instead of just one single value when it is queried. In other words, when a node requests time service by calling TT.now(), it returns an interval [earliest, latest], where both 'earliest' and 'latest' are timestamps and the absolute time when this call happens is represented by a timestamp inside this interval. Based on this time service, external consistency is preserved: if the actual starting time of a Read-Write transaction T2 happens after the actual commit time of Read-Write transaction T1, T2's commit timestamp is greater than that of T1's. Such an implementation turns out to be serializable and the proof goes as follows.

        Commit timestamp s1 for T1 

        < Absolute time of commit event(the start of second phase of two-phase commit) of T1

        < Absolute time of releasing the conflicting lock for T1

        < Absolute time of acquiring the conflicting lock for T2

        < Absolute time of requesting TT.now() when T2 receives the commit request

        <  Commit timestamp s2 for T2                                                                                                  
	  

	                                                                                                       ## 
	  

The so-called absolute time in [Co 13] is just a synonym of the real clock time of the reference frame used throughout this article. I started looking for conflict loops in histories in Google's Spanner and couldn't find any. After a while I realized that this hybrid isolation level might be a serializable one and tried to give a proof. It was still a surprise to me after I proved it since I thought a merge of 2PL and Snapshot Isolation would only give us an isolation level at least as good as Snapshot Isolation, not necessarily the highest one. I don't know if the developers of Spanner know that their database actually implements the serializable isolation level. It is not mentioned in [Co 13].

If an application in Spanner contains datetime related fields, we might have to apply a version of the Ramification Theorem to conclude Serializability on Ram(B)', B being the set of datetime fields, since I am not sure the timestamp system they use can get rid of all the timestamp-based inconsistency.

A row in a Spanner table consists of three parts: a primary key, the commit timestamp of this row as a version number, the data. So it is practically a key-value data-store with versions. Chances are that a variant of the proof in Example 12 can be used to show that Spanner is serializable. But I still like this proof better since it is more generic. It is a pity that Spanner didn't implement a full-blown relational database, otherwise we may see this serializability implementation's full power in action. Wish we could see that in the future.

D.2 OceanBase

OceanBase is a product of Alibaba. It seems to me it borrows a lot of wisdom from Oracle, in that it implements a Snapshot Isolation a lot like Oracle's, except that it is a distributed system; and some wisdom from Google's Spanner, in that it uses a variant of Paxos system called Raft to replicate data too. OceanBase's implementation of Snapshot Isolation is different from that of Google's Spanner's in that it uses a machine generated increasing sequence, instead of a clock, to timestamp the transactions. For a standalone system, the generation of such a sequence is straight forward. But OceanBase is a distributed system, the increasing sequence must be generalized to a global version. OceanBase deploys specific servers, so called Global Timestamp Service(GTS) in its nomenclature, for that purpose. GTS provides an increasing sequence for timestamp purpose for each Tenant(synonym of database instance in OceanBase's nomenclature, roughly corresponds a store in their Amazon.com like on-line store application called Tmall, usually consists of one leader and two replicas, located on three machines) and there are as many of these sequences as there are Tenants. Such a generalization offers about two million timestamps per second for each Tenant, which is usually much more than a store's need. Then for only transactions accessing the same Tenant, Snapshot Isolation is used to provide consistency. In other words, although Oceanbase's architecture is a distributed one, the unit that implements Snapshot Isolation is really standalone, namely, in the leader of a tenant. One advantage of using an increasing sequence to provide timestamp service is that realizing 'external consistency' becomes trivial.

This partition-like implementation of Snapshot Isolation, however, imposes a major limitation to OceanBase: there is NO consistency support whatsoever for an application that consumes more than two million timestamps per second unless it can be partitioned into a lot of Tenants like an Amazon.com application. In other words, if consistency is important for such an application, you have to implement it at the second tier. Believe me, that usually implies heavy lifting! Google's Spanner uses a bunch of servers that are synchronized regularly to provide timestamp service and you can query any of them, so the limit should be that of the whole bunch and the danger for them to become saturated is less severe. And the method I've developed for NDB Cluster doesn't exhibit this kind of limitation at all since it doesn't require timestamps to achieve consistency. It seems to me OceanBase is designed for an application like Amazon.com and that is it; for other application types, it has to fit into a single Tenant. Such a design of course is suitable for a cloud environment since a tenant may not occupy a whole machine.

Besides this limitation of the timestamp service, OceanBase also suffers the same problems Snapshot Isolation does, such as write-skew anomalies and cascading abortion. That said, OceanBase has survived Alibaba's annual sale on Nov. 11th(an online version of Black Friday Sale). It also occupies the first place in the TPCC benchmark list for clusters. This is impressive, especially if you are building an application like Amazon.com. Of course, if in your application SQL statements are executed upon data that span across multiple data nodes(say, data can't fit in a single data node), this benchmark result may not be meaningful for you since OceanBase doesn't support this kind of statements naturally due to its limitation. Its timestamp service, however, doesn't require hardware provision like GPS clocks and atomic clocks that Google's Spanner does and hence its deployment could be more cost-efficient when it applies.

We've talked about external consistency about Spanner and OceanBase. Let's take a look at NDB Cluster. External consistency demonstrated in Spanner and OceanBase doesn't apply to NDB Cluster since it is not a timestamp-based system. There does exist the following abnormal behavior: for two sequentially started Read-Only transactions, the first one return a new value written by a Read-Write transaction because of a delay(for example, the thread handling this transaction got context-switched out) while the second one gets an old value. External consistency is about whether a client of the application feels right. What we emphasize in this article is to make the database right. And we've built a serializable system such that consistency based data corruption is managed, even if a client feels odd sometimes.

D.3 CockroachDB

CockroachDB implements a serializable isolation level and the following analysis is based on [Tr 16].

CockroachDB is a distributed database system that uses a global clock called Hybrid Logical Clock(HLC) as described in [Ku 14] for its serializability implementation. HLC is a combination of the logical clock provided by Lamport in [La 78], and the physical clock provided by NTP. Each timestamp is in the form of (l, c), where l is the timestamp returned by NTP and c is a counter. The primary algorithm for maintaining such a clock is given as follows:

      Initially l.j := 0; c.j := 0

      Send or local event
        l'.j := l.j;
        l.j := max(l'.j, pt.j);
        If (l.j=l'.j) then c.j := c.j + 1
            Else c.j := 0;
        Timestamp with l.j, c.j

      Receive event of message m
        l'.j := l.j;
        l.j := max(l'.j, l.m, pt.j);
        If (l.j=l'.j=l.m) then c.j := max(c.j, c.m)+1
            Elseif (l.j=l'.j) then c.j := c.j + 1
            Elseif (l.j=l.m) then c.j := c.m + 1
            Else c.j := 0
        Timestamp with l.j, c.j                                                                                                     
	

Here sub-script j represents an arbitrary server participating in this clock protocol, sub-script m represents the server from which the message(also represented as m) is sent, and pt represents the physical time returned by the NTP protocol. As we can see, this clock protocol stores the latest physical clock it can sense in l, which is usually enough to describe the clock advancement; if it is not, c is there for that purpose. Notice that such a clock deviates from a traditional one, which usually assigns a number to each event. The order between the timestamps is the lexicographic order, which is a total order. And in particular, we have

      (l1, c1) < (l2, c2) iff l1 < l2 or l1 = l2 and c1 < c2.                                                                                                    
	

With e and f representing events in the system, this clock demonstrates the following nice properties:

1. e 'happened before' f => (l.e, c.e) < (l.f, c.f).
2. l.e is close to pt.e, i.e., |l.e – pt.e| is bounded by Epsilon, where the real clock time lies within the interval [pt.e – Epsilon/2, pt.e + Epsilon/2].
3. (l, c) has O(1) computational complexity and space requirement.

Remark: For the “happened before” partial order in [Ku 14], we can't just use the one described in [La 78] since [La 78]'s is for a system with single-threaded processes and today's systems are almost always multi-threaded. We should use the variant where events inside a thread are represented as points in time and two events connected by an inter-thread message is interpreted as a type II edge as mentioned in Appendix B. If we were to use the variant 'happened before' as in the rest of this article, some of the results in [Ku 14], like theorem 4, may not be easily proved. Notice that using this “happened before” variant doesn't affect the correctness of the serializability theorems since it is only used in the timestamp service in of the system; for the serializability theorems, we still use 'happened before' as in the rest of this article. This is a perfect example where we may interpret the temporal order in a distributed system in different ways mathematically.

Property 1, the so-called 'Clock Condition' in [La 78], is extremely important, for example, if we were to use it to replace the expansive distributed timing service in Google's Spanner to generate a consistent snapshot for the 'optimistic' component.

Property 2 suggests that this clock heavily depends on the physical clock provided by NTP. On the one hand, this means things like violation of external consistency becomes less visible for clients if NTP time is synced to be close to real clock time; on the other hand, this also implies if NTP synchronization is seriously off, it might impact this clock protocol.

For example, if topology of the network allows a server to be disconnected from a NTP source, but remains connected to other participants of the clock protocol, this server's physical time will drift away from NTP time if the clock onboard this server is skew. [Ku 14] claims that they can handle unstable situations like this with two supplementary rules to the primary algorithm: reset the participant's l to pt(the time provided by the physical clock onboard the server, not NTP time) and start again, and ignore out-of-bounds messages. With the first rule, [Ku 14] claims in the case of extreme clock errors by NTP or transient memory corruption, HLC can be corrected once physical/NTP clock stabilizes; with the second rule, [Ku 14] not only ignores out-of-bound messages, but also increases c by 1 when that happens. Both rules deviate from the primary algorithm and the question is whether the properties still hold with these modification. [Ku 14] doesn't contain a justification to this.

What is more, in property 3, the c value is actually bounded from above by a function of Epsilon. It would be desirable for Epsilon to remain a constant when the clock protocol is operating. For example, we probably allocate a large enough integer variable for c according to this upper bound. If Epsilon fluctuates, the variable might overflow. An erroneous state of the NTP protocol might give us exactly that.

It seems the developers of CockroachDB seem agree with me on this. They deal with it with the following rule: when a node detects that its clock is out of sync with at least half of the other nodes in the cluster by 80% of maximum offset allowed, it crashes immediately. This approach pertains to the primary algorithm all the time, although it might imply the algorithm operates with a larger Epsilon sometimes(it really depends on the relation between Epsilon and the maximum offset). This implies any property that doesn't involve Epsilon, in particular property 1, remains true. So if we were to use HLC for an implementation of Serializable Snapshot Isolation or Spanner's consistent snapshot, they will both remain correct with unstable situations like drift-away servers since it is equivalent to a clock protocol that operates with a larger Epsilon. And the issue described in the last paragraph will also go away if we choose the variable for c carefully based on the maximum offset instead.

There are two consequences with CockroachDB's approach. One is that anomalies like violation of external consistency will show up more often when the clock protocol operates with a larger Epsilon. The other is that if HLC co-locates with data, the brought-down of the server could render those data unavailable(unless there is a replication of that data like in NDB Cluster). This could further compromise the semantics of the deployed application(For example, if the average of a field is computed system-wide in the application, it might render it biased if part of the data is not available). Notice that Google's Spanner is free from this issue since they use dedicated servers for their TrueTime service.

Now let's examine CockroachDB's serializability implementation. In the arena of concurrency control, it turns out besides 'optimistic technology' and 'pessimistic technology', there is a third species called 'timestamp-based technology'. CockroachDB's serializability implementation is a typical one of it and of course the timestamps in it are provided by HLC.

When a transaction starts, it is assigned a timestamp. Every operation in that transaction is considered to take place at that timestamp. The following three rules are to handle WR, RW and WW conflicts. Let's assume the conflict in concern is from transaction T1 to T2, or T1 → T2, as usual and the associated timestamps are t1 and t2 respectively. And when we say a read, we refer to either an item read or a predicate read.

WR conflicts: CockroachDB implements a multi-version database. When a read happens in T2, the system returns the most recent version with a timestamp t < t2. Hence whenever there is a WR conflict between T1 and T2, t1 < t2.

RW conflicts: Upon any read operation, the timestamp is recorded in a node-local timestamp cache. This cache will return the most recent timestamp at which the key is read when queried. All write operations query the timestamp cache for the key they are writing. If the returned timestamp is greater than or equal to the timestamp of the write operation, this writing transaction is aborted and will be restarted with a later timestamp. Now suppose t1 >= t2, then when T1 performed its read, T2 hadn't perform its write yet since otherwise T1's read would see a version greater or equal to that established by T2 and contradicts with the fact that the conflict is a RW one. When T2 performed its write after T1's read, the cache would return at least t1. This would cause T2 to abort and be restarted with a timestamp larger than t1. This is a contradiction. Hence if the conflict between T1 and T2 is a RW one, we also have t1 < t2. Notice in a RW conflict, the read doesn't necessarily 'happen before' the write. One can find example like this easily in Snapshot Isolation.

WW conflicts: If a write operation attempts to write to a key, but that key already has a version with a timestamp greater than or equal to that of the operation itself, the transaction of this operation is aborted and will be restarted with a later timestamp. Again if the conflict between T1 and T2 is a WW one, we have t1 < t2.

Notice that in the WR case, the read refers to both item read and predicate read. In the RW case, the timestamp cache is an interval cache and the key range involved in the read is recorded, hence both item read and predicate read are taken care of. We've just demonstrated a history in this system is free of conflict loops since < is a strict partial order. So as long as CockroachDB interprets predicate-based conflicts correctly as in the Serializability Theorems, this should provide a serializability implementation.

Unfortunately CockroachDB, like PostgreSQL, also resorts to [Be 87] for its serializability proof, which has been demonstrated to be unsuitable for modern workloads where secondary key accesses are ubiquitous. Of course, it is up to CockroachDB to clarify this.

There is another issue with this serializability implementation, related to the rule about RW conflicts. It could actually result in a starvation situation for a write operation if too many reads are intended for the same data item. Consider the following situation: a Read-Write transaction is computationally intensive(hence long running) and the result is updated to a field(say, we are computing the Nasdaq index and we need to query a lot of fields and do some serious computation on them), and a lot of short-lived Read-Only transactions constantly floods in to read this field since it is critical for business. This way, chances are that when the Read-Write transaction finishes, the cache already records a lot of reads of that field with timestamps larger than that of the Read-Write transaction's and it has to be rolled back and restarted. The semantics of this application is certainly for this field to be updated as soon as it is available and this starvation situation does exactly the opposite of that. Notice this starvation situation is worse than the case where a write lock waits for a bunch of read locks since in the later case the write will eventually be served. This is not a surprise since the rule for RW conflicts basically says that a read with a later timestamp has higher priority than a write on the same item with an earlier timestamp. Usually we want an update be reflected in the database ASAP. CockroachDB sacrifices this for serializability, but it turns out there is a price.

Remark: The first time I encountered a timestamp-based serializability implementation was in [Si 02] when I explored technical paths of achieving serializability more than a decade ago. Comparing with that, there are two advancements in CockroachDB's implementation. The first one, of course, is the adoption of distributed timestamp service HLC so that serializability can be discussed in a distributed database system. The second one is the use of a multi-version system so that transaction with an earlier timestamp than that of the field it's trying to read doesn't have to be aborted. The starvation issue was also present in [Si 02]'s implementation. I couldn't resolve it and that was one major reason I turned to 'optimistic technology' and 'pessimistic technology' for a solution. Today I still can not. I am open to any suggestion to resolve it to make this implementation more sensible.

D.4 TiDB

I'd like to explicitly thank 'Hazel' and her colleague from TiDB for answering my questions about TiDB's timestamp service to begin this subsection.

TiDB is distributed, disk-based database system. TiDB claims it implements two isolation levels: Repeatable-Read and Read-Committed. It actually implements three: with isolation level setting to Repeatable-Read and system configuration variable tidb_txn_mode setting to 'optimistic', we got a Snapshot Isolation implementation; with isolation level setting to Repeatable-Read and system configuration variable tidb_txn_mode setting to 'pessimistic', we got a Repeatable-Read isolation level that is similar to that of MySQL InnoDB's; with isolation level setting to Read-Committed and system configuration variable tidb_txn_mode setting to 'pessimistic', we have a Read-Committed isolation level that is similar to that of MySQL InnoDB's.

For the audience of this article, Snapshot Isolation may not be interesting; but the distributed clock used to implement it is. We'll explore it in detail here. Let's start with TiDB' architecture.

The architecture of TiDB is described here. There are three kinds of server in a TiDB cluster: TiDB server, Placement Driver(PD) server and TiKV server. A TiDB server is like a MySQL server in a NDB Cluster which serves as an API server to interface the clients. The PD servers serve as the heart of the cluster and the collection of them is like the NDB kernel in a NDB Cluster; among other things, the timestamp service of the distributed clock is provided by them. The TiKV servers are responsible for storing data, just like the data nodes in a NDB Cluster.

A TiKV server actually uses RocksDB to store its data to local storage. RocksDB is a key-value store, hence the name TiKV for such a server. Although RocksDB is a key-value store, TiDB is a full-blown relational database. Every row in a TiDB table is transform into a key-value pair before it is stored in the underlying RocksDB. For example, in the primary index of a table with four columns col1, col2, col3 and col4, a row is represented as

      key:    tablePrefix{TableID}_recordPrefixSep{RowID}
      value: [col1, col2, col3, col4]                                                                                               
	

in RocksDB. A similar scheme applies to secondary indices. Details can be found here.

TiDB also provides high availability with a variant of Paxos system called Raft, just like OceanBase does. The timestamp service, for example, is located in the leader of a Raft group formed by PD servers. This timestamp service is used to provide two timestamps for each transaction to signify the start and commit of it so that Snapshot Isolation can be implemented. This timestamp service is the most interesting part of this Snapshot Isolation implementation. Let's see how it overcomes OceanBase's issue of NOT being able to provide sufficient number of timestamps to the whole cluster with an increasing sequence.

The timestamp service in TiDB is called Timestamp Oracle(TSO). The object for defining it is as follows:

      type tsoObject struct {
        physical time.Time
        logical int64
        …
      }                                                                                          
	

Here 'physical' represents a Unix timestamp in milliseconds which is 46-bit long, and 'logical' is an 18-bit number trying to provide as many as 2^18 timestamps per millisecond. When a client requests a timestamp, what it got is a pair of integers in the following form: (physical, logical). The component responsible for responding to such a request is called an allocator and it resides in the leader of the Raft group. When the physical field is NOT updated, the allocator only returns a bunch of timestamps with that fixed physical field and various logical fields in an increasing order upon requests. The update of the physical field also generates an increasing sequence in its own and it is the most intricate part of the algorithm. The operation of this part of the timestamp service is explained here.

We need four variables for this part of the algorithm:

      Tlast: The physical field stored, on disk or in the TSO object.

      Tnext: Next value for the physical field.

      Tnow: Time returned by the Wall clock.

      Tx: Duration of each round of timestamp service, defaults to 3 seconds.                                                                                  
	

A Raft group leader for the PD servers is elected at cluster's initial start or re-elected after the previous leader crashes. This new PD Leader then starts the calibration of the physical time. It first retrieves Tlast persisted by the previous PD Leader from the disk. This Tlast value represents the upper bound of previous round of timestamp service for the physical field and hence every timestamp serviced before the crash got a smaller physical field than Tlast. Then the new PD compares Tlast with Tnow:

      If Tnow - Tlast < 1 ms, then Tnext = Tlast + 1; otherwise Tnext = Tnow.                                                                                 
	

Then Tnext is stored in the physical field of the TSO object and this finishes the calibration of the time service. Before the allocator starts to service timestamp, Tnext + Tx is stored to disk through Raft, so that it could survive the next crash and again be retrieved as Tlast upon restart. Also all timestamps in the time period [Tnext, Tnext + Tx)(3000 x 2^18 of them) are generated in the memory. Then the allocator starts shoveling out timestamps upon request.

Every 50 ms, the algorithm tries to advance the physical field of the TSO object so that the service won't run out of the 2^18 allotment for the logical field. This process is called a Fast-Forward and it works like this:

It first calculates JetLag := Tnow – Tlast, here Tlast is the value retrieved from the TSO object. If JetLag > 1ms, the physical time in the hybrid logical clock is slower than the current physical time, and needs to be fast-forwarded to make Tnext = Tnow.

In addition, when the logical field reaches its threshold value(2^18) during the time services, it stops and waits for the next Fast-Forward. So, to prevent this situation from happening, the Fast-Forward process also checks if the current logical field exceeds half of its threshold value, the physical field in the TSO will be advanced as well if it is affirmative. In this case, Tnext = Tlast + 1.

Then Tnext is stored in the physical field of TSO, where its logical field is set to 0 and the timestamp service resumes after Fast-Forward.

In the Fast-Forward process, the logic also checks whether Tnext is too close to Tlast, the value stored on disk. If the answer is affirmative, a new Tlast = Tnext + Tx is generated and stored on disk so that the timestamp service can continue. The source code for the allocator and TSO are here and here(mainly in the syncTimestamp() and updateTimestamp() functions).

From this description of the timestamp service algorithm, we can see that an allocator hands out timestamps in an increasing order. So naturally we would like to compare it with OceanBase's timestamp service which provides an increasing sequence by a thread for each tenant in the cluster. Each second, TiDB's timestamp service can generate 1000/50 x 2^18 = 5.2M timestamps, which is more than twice that of OceanBase's. Besides pre-allocating all the timestamps in memory, TiDB also services timestamps in a batch. These are two reasons why TiDB's timestamp service trumps.

The 50ms duration between two consecutive Fast-Forwards is the default value of the 'tso-update-physical-interval' configuration parameter, which can be set to a minimum value of 1 ms for more timestamps at the cost of 10% more CPU usage. So each second, the timestamp service can theoretically generate a maximum of 10^3 x 2^18 = 2.6 x 10^8 timestamps. If this theoretical limit can actually be reached, the timestamps generated are two order of magnitude of that of OceanBase's. However, the default value could be an empirical sweet spot, so please make sure you test it out if you want to alter it.

With the default setting of 'tso-update-physical-interval', the maximum throughput a TiDB cluster can support is about 2.6 x 10^6 tps since each transaction requires two timestamps. If this is not enough for your deployment, TiDB supports a service mode that is similar to OceanBase's: it may provide a key space, which is like a tenant in OceanBase, for each application; a TSO microservice(with one allocator) is split from PD services to service each key space group; in the extreme case, there could be only one key space in such a group and this replicates the OceanBase case – one increasing sequence as timestamps for each tenant. This is a new feature that is available starting with version 8.0.0.

With this distributed clock, TiDB provides a Snapshot Isolation implementation for its OLTP workloads. However, this distributed clock doesn't satisfy the Clock Condition and hence we can't apply the work in Section 5 to it directly to achieve serializability, although we've come to the conclusion that we may also apply type B of the Serializability Theorem to generalize Serializable Snapshot Isolation to a distributed Snapshot Isolation implementation with tuple versions there.

The fact that it doesn't satisfy the Clock Condition is easy to observe: suppose in node A , transaction T1 updated a tuple and a request was sent to node B to perform a read, then it requested the commit timestamp s1; meanwhile, T2 in node B requested the commit timestamp s2 after node B had serviced T1's read; if the Clock Condition was honored, then we should have s1 < s2; but since TiDB uses batching to request timestamps, if the batch in node A was empty when s1 was requested while node B's batch was almost full, then s1 > s2 was possible since node A's batch could be sent out much later than node B's. Even if we could turn off timestamp request batching, a similar situation could still arise because TiDB supports geo-distributed clusters and WAN provides a lot of uncertainty, including node A's request arriving late at PD.

So if we still like to apply the work in Section 5 to conclude serializability in TiDB, the proof to Theorem 2.1 in [Fe 05] in subsection 5.1 must be done again with TiDB's timestamp service. Fortunately, there are only two places we need to modify in that proof.

The first one is that we used the Clock Condition to conclude that T's start timestamp < T's commit timestamp for every transaction Tin the proof. And this can be seen from TiDB's optimistic transaction model here: both timestamps are requested by the TiDB server that is coordinating the transaction and the commit timestamp is requested after the start timestamp's arrival; hence the returned start timestamp will be smaller than the commit one since TSO returns timestamps in an increasing order.

The other one concerns the WW conflict case in the proof of Lemma 2.2 in [Fe 05] and I am quoting the arguments in subsection 5.1 here:

'If the conflict between T1 and T2 is a WW one, by First-Committer-Wins rule or First-Updater-Wins rule, we have timestamp of T1's start < timestamp of T1's commit(Clock Condition) < timestamp of T2's start < timestamp of T2's commit(Clock Condition). Notice the second inequality can't be the other way around(timestamp of T2's commit < timestamp of T1's start) since then we would have the following timestamp loop: timestamp of T2's commit < timestamp of T1's start < timestamp of T1's conflicting write < timestamp of T2's conflicting write < timestamp of T2's commit by the Clock Condition.'

In the TiDB case, the inequality timestamp of T1's commit < timestamp of T2's start can be shown by the logic of write conflict resolution here: when transaction T2 tries to commit, it checks whether the condition data.commit_ts > T2.start_ts is true; if it is affirmative, some other transaction commits a version with a larger timestamp and T2 is rolled back; either T2 is rolled back or not, we have data.commit_ts < T2.start_ts because TSO returns timestamps in an increasing order; but of course, data.commit_ts belongs to transaction T1.

Notice that although TSO also uses physical time in its design, it seems the caveats about using physical time discussed thoroughly in Appendix B and the subsections about OceanBase and CockroachDB in Appendix D are masked up by the logical part, at least for the key property that TSO returns timestamps in an increasing order, which is crucial for the previous arguments to go through.

Now that we've proved Theorem 2.1 in [Fe 05] with the timestamp service of TiDB, we may apply it to conclude that if the DSG of an application is free of 'dangerous structure', it is serializable. In the case of TPCC, from the analysis in Example 16, we may sketch the DSG as in figure 13 of [Fe 05] and see that there are two possible 'dangerous structure's.

The first one is from the following RW conflicts: the first or fourth RW conflict in the s → n case, and the second RW conflict in the n → n case. However, this 'dangerous structure' is not real. That is because for the RW conflict in the n → n case to be possible, the two new-order transactions must also update the s_quantity field of the same tuple. This would push them away and prevent them to be concurrent under Snapshot Isolation and the possible RW conflict would disappear too. Notice in a Read-Committed like that of NDB Cluster's, it is possible for these two transactions to be concurrent.

The second one is from the following RW conflicts: one of the three RW conflicts in the o → d2 case, and one of the first three RW conflicts in the d2 → d2 case. But from the discussion in the d2 → d2 case, we know that the latter RW conflict doesn't exist under Snapshot Isolation, in particular TiDB's. Hence this 'dangerous structure' is not real either. Again, in a Read-Committed like that of NDB Cluster's, it is possible for these two transactions to be concurrent. So now we can conclude serializability for TPCC in TiDB's Snapshot Isolation.

In the case 'dangerous structure's do exist, we need to use techniques like materialization or promotion to eliminate them. Please refer to [Fe 05] for an exposure of these techniques. Notice that joins need to be re-written as equivalent statements as in the serializability implementation I've developed for NDB Cluster before conflict analysis.

We may also apply the method I've developed in Section 1 to achieve serializability to this specific Snapshot Isolation implementation with materialization and promotion since it can be viewed as a generic Read-Committed isolation level. But it is apparently inferior comparing with Fekete's method, so I won't elaborate on it any more.

And of course you may apply the Generalized Serializability Theorems to enhance performance while guaranteeing consistency of the underlying database, if inconsistencies in some reads are OK; in that case, please use the conflict_parser helper program to get you started analyzing the conflicts. Or you may use the Ramification Theorems to constrain inconsistencies to Ram(B). Please refer to Example 21 for a demonstration of the calculation of Ram(B).

A transaction under TiDB's Snapshot Isolation reads the most recently committed tuple version is guaranteed by the following arguments: a transaction T1 reading at timestamp s1 and a transaction T2 that committed at timestamp s2 < s1; we will show that T1 sees T2’s writes. Since s2 < s1, we know that the TSO gave out s2 before or in the same batch as s1; hence, T2 requested s2 before T1 received s1. We know that T1 can’t do reads before receiving its start timestamp s1 and that T2 wrote locks before requesting its commit timestamp s2 . Therefore, the above property guarantees that T2 must have at least written all its locks before T1 did any reads; T1’s Get() will see either the fully committed write record or the lock, in which case T1 will block until the lock is released. Either way, T2’s write is visible to T1’s Get(). These arguments can be found here. Notice TiDB provides this guarantee with the following key properties of the system: monotonicity of timestamp service, replication of locks for optimistic transaction by the Raft system, reads waiting for locks to be released.

Durability of consistency is guaranteed by Theorem 6 since durability is provided by Raft.

Unfortunately, we can't mount a serializability implementation at 2nd-tier to the two pessimistic isolation level implementations because a tiny piece of the puzzle is missing. In the successful mounting of a serializability implementation to a NDB Cluster, the key part is the proof of Theorem 1 & 2. If we choose the commit event to be one that is after execution of all SQL statements, but before 2PC described here, we can see that the proof of Theorem 1 and the 'If' part of Theorem 2 can go through in TiDB's pessimistic isolation level implementations, but not the 'Only If' part. The 'Only If' part depends on Example 14 and in the code snippet similar to the following one there, the locking read in T1 blocks for T2's insert in NDB Cluster for the arguments to be correct.

                              T1                                                      T2

                                                                                start transaction;

                                                                                insert into t values (2,2,2,2);

      start transaction;

      select * from t where id1>0 and id1<4 for update;                                                                           
	

But in TiDB, T1 doesn't block for T2 and the arguments in the 'Only If' part of Theorem 2 can't go through. Let's see a possible scenario in TiDB that this behavior would render the 'Only If' part to be incorrect. Suppose T1 and T2 are defined as follows:

                  T1

      start transaction;

      acquire l1;
    
      select * from t where id1>0 and id1<4 for update;

      commit;
    
                  T2                                 

      start transaction;

      insert into t values (2,2,2,2);

      acquire l2;
                                                             
      commit;                                                                            
	

Here l1 and l2 are conflicting locks as in Theorem 2.

Consider the situation when T2 is committing and 2PC starts to release locks acquired. In this releasing process, there is a chance that l2 has already been released, but the write lock for the insert hasn't. If this status lasts long enough, T1 may acquire l1 and start the execution of the locking read. Since the insert doesn't block this locking read, it will read the the database state before the insertion. So we have a predicate RW conflict between T1 and T2 and yet l2 → l1. This refutes the 'Only If' part of Theorem 2.

Finally, let's round up this subsection by comparing it with NDB Cluster in the following table:

* Region balancing is explained here.

** TiDB currently doesn't implement gap locking. But gaps within the same region are not destroyed.