Transaction Semantics in Apache Kudu

Kudu employs Multiversion Concurrency Control (MVCC) and the Raft consensus algorithm [4]. Each write operation in Kudu must go through the tablet’s leader.

All replicas of a tablet observe the same order of operations, and if a write operation is assigned timestamp n and changes row x, a second write operation at timestamp m > n is guaranteed to see the new value of x.

This strict ordering of lock acquisition and timestamp assignment is enforced to be consistent across all replicas of a tablet through consensus. Therefore, write operations are totally ordered with regard to clock-assigned timestamps, relative to other writes in the same tablet. In other words, writes have strict-serializable semantics, though in an admittedly limited context. See this blog post for a little more context regarding what these semantics mean.

While Isolated and Durable in an ACID sense, multi-row write operations, even within a single tablet, are not fully Atomic unless they are a part of a multi-tablet write transaction. The failure of a single write in a batch operation does not roll back the operation, but produces per-row errors.

Regardless of whether they are a part of a transaction, writes from a Kudu client are optionally buffered in memory until they are flushed and sent the server. When a client’s session flushes, the rows for each tablet are batched together, and sent to the tablet server that hosts the leader replica of the tablet. Outside of a transaction, each of these batches represents a single, independent write operation with its own timestamp. However, the client API provides the option to impose some constraints on the assigned timestamps and on how writes to different tablets are observed by clients.

Kudu, like Spanner, was designed to be externally consistent [5], preserving consistency when operations span multiple tablets and even multiple data centers. In practice this means that, if a write operation changes item x at tablet A, and a following write operation changes item y at tablet B, you might want to enforce that if the change to y is observed, the change to x must also be observed. There are many examples where this can be important. For example, if Kudu is storing clickstreams for further analysis, and two clicks follow each other but are stored in different tablets, subsequent clicks should be assigned subsequent timestamps so that the causal relationship between them is captured.

Kudu’s default external consistency mode is called CLIENT_PROPAGATED. See [1] for an extensive explanation on how it works. In brief, this mode causes writes from a single client to be automatically externally consistent. In the clickstream scenario above, if the two clicks are submitted by different client instances, the application must manually propagate timestamps from one client to the other for the causal relationship to be captured.

Timestamps between clients a and b can be propagated as follows:

Kudu also has an experimental implementation of an external consistency model used in Google’s Spanner , called COMMIT_WAIT. COMMIT_WAIT works by tightly synchronizing the clocks on all machines in the cluster. Then, when a write occurs, timestamps are assigned and the results of the write are not made visible until enough time has passed so that no other machine in the cluster could possibly assign a lower timestamp to a following write.

When using this mode, the latency of writes is tightly tied to the accuracy of clocks on all the cluster hosts, and using this mode with loose clock synchronization causes writes to take a long time to complete or even time out. See Known Issues and Limitations.

The COMMIT_WAIT consistency mode may be selected as follows:

Kudu provides transactionality on top of the write operations, meaning all operations that occur within a transaction abide by the same consistency behavior described above.

When a client begins a transaction, Kudu automatically assigns the transaction a unique identifier (called a "transaction ID"). The identifier can be used to create sessions to which write operations are applied, potentially across multiple clients per transaction. Write operations applied in the context of a transaction are not visible until a client commits the transaction.

Kudu exposes the following APIs to pass a transaction identifier between potentially multiple processes:

As writes are applied in the context of the transaction, each tablet that participates in the transaction automatically registers itself as a participant, and is locked for further transactions until the transaction is complete. Per-row locks are taken as per the normal flow of a write operation, but per row locks are released upon replicating the write operation, in favor of relying on the tablet-wide lock.

If multiple transactions lock the same tablet, Kudu uses the wait-die scheme to avoid deadlocks when locking the participant: if a transaction b attempts to lock a tablet that is already locked by transaction a, if a > b (a is newer than b), transaction b continues trying to lock until it is successful (it "waits"). Otherwise, transaction b is automatically aborted, and it is up to the application to retry the transaction.

When the client commits a transaction, Kudu orchestrates a two-phase commit that assigns a "commit timestamp" to all write operations that is higher than each of their individually assigned timestamps. The mutations of the transaction are all visible to clients as of this commit timestamp. Additionally, subsequent write operations on all participants are guaranteed to be assigned timestamps higher than this timestamp. It is up to applications to ensure that all desired write operations have succeeded (i.e. did not return row errors) before committing.

As long as a transaction is expected to remain active, applications are expected to maintain at least one reference to the given transaction’s handle, each of which can be configured to automatically heartbeat to the Kudu cluster, indicating liveness of the transacting application. By default, only the first created transaction handle for a transaction will heartbeat, with the expectation that it is kept alive for the entire duration of the transaction. If only a single transaction handle is expected to be kept alive at once across multiple clients, the heartbeating can be enabled with the following calls when serializing the handle for use in other processes.

Scans are read operations performed by clients that may span one or more rows across one or more tablets. When a server receives a scan request, it takes a snapshot of the MVCC state and then proceeds in one of two ways depending on the read mode selected by the user. The mode may be selected as follows:

The following modes are available in both clients:

Selecting between read modes requires balancing the trade-offs and making a choice that fits your workload. For instance, a reporting application that needs to scan the entire database might need to perform careful accounting operations, so that scan may need to be fault-tolerant, but probably doesn’t require a to-the-microsecond up-to-date view of the database. In that case, you might choose READ_AT_SNAPSHOT and select a timestamp that is a few seconds in the past when the scan starts. On the other hand, a machine learning workload that is not ingesting the whole data set and is already statistical in nature might not require the scan to be repeatable, so you might choose READ_LATEST instead for better scan performance.

There are several gaps and corner cases that prevent Kudu from being fully strictly-serializable in some situations, at the moment. Below are the details and next, some recommendations.