KeyValue explicit timestamps vs memstoreTS

In this post I tried to summarize HBase's ACID implementation. This topic was also the main subject of my talk at HBaseCon.

I talked to a few folks afterwards, and it seems there is some general confusion about the memstoreTS (and MVCC) and how it relates to the explicit timestamp stored with each KeyValue.


The key difference is that the timestamp is part of the data model. Each KeyValue has a rowkey, a column identifier, a value... and a timestamp.
An application can retrieve and set the timestamp, decide as of which timestamp(s) it wants to see a certain value, etc. It should be considered as just another part of a column (a KeyValue pair to be specific) that happens to have special meaning.

The memstoreTS on the other hand is a construct purely internal to HBase. Even though it is called a timestamp (TS), it has nothing to do with time, but is just a monotonically increasing "transaction number". It is used to control visibility (MVCC) of a "transaction" in HBase (i.e. a Put or Delete operation), so that changes are not partially visible.

Remember that the explicit timestamp is part of the application data that gets changed by a transaction.

Say you have two concurrent threads that update the same columns on the same row (a row is the unit at which HBase provides transactions):
Thread1, Put: c1 = a, c2 =b
Thread2, Put: c1 = x, c2 =y

That's cool. One of the threads will be first, HBase makes sure (via the memstoreTS) that Puts are not partially visible, and also makes sure that all explicit timestamp are set to the same value for the same Put operation.
So any client (looking at the latest timestamp) will either see (a,b) or (x,y).

Things get more confusing when a client sets the explicit timestamp of (say) a Put. Again consider this example where two concurrent threads both issue a Put to the same columns of the same row, but set different timestamps:

Thread1, Put: c1(t1) = a, c2(t2) =b
Thread2, Put: c1(t2) = x, c2(t1) =y
(t2>t1)

So what will a client see?

The first observation is still that each Put is seen in its entirety, or not at all.

If a Get just retrieves the latest versions and does so after both Thread1 and Thread2 finished it will see:
c1(t2) = x, c2(t2) = b

This is where many people get confused.

It looks like we're seeing mixed data from multiple Puts. Well, in fact we do, but we asked for only the latest versions, which include values from different transactions; so that is OK and correct.

To understand this better let's perform a Get of all versions. (i.e. call setMaxVersions() on your Get). Now we'll potentially see this:

• nothing (if the Get happens to be processed before either the Puts).
• c1(t1) = a, c2(t2) = b (Thread1 was processed first, and Thread2 is not yet finished)
• c1(t2) = x, c2(t1) = y (Thread2 was processed first, and Thread1 is not yet finished)
• c1(t1) = a, c1(t2) = x, c2(t1) = y, c2(t2) = b (both Thread1 and Thread2 finished)

We see that each Put indeed is either completely visible or not at all, and that HBase correctly avoid visibility of a partial Put operation.
In addition we can confirm that the timestamp is part of the data model that HBase exposes to clients, and the transaction visibility has nothing to do with the application controlled/visible timestamps.

Lastly in this scenario we also see that it is impossible for a client to piece together the state of the row at precise transaction boundaries.
Assuming both Thread1 and Thread2 have finished, a client can request the state as of t1 (in which case it gets c1(t1) = a, c2(t1) = y) or as of t2 (in which case it'll see c1(t2) = x, c2(t2) = b).

This behavior is the same as you'd expect from a relational database - the current state is the result of many past transactions; with the extra mental leap that the timestamp is actually part of the data model in HBase.

HBase client timeouts

The HBase client is a somewhat jumbled mess of layers with unintended nested retries, nested connection pools, etc. among others. Mixed in are connections to the Zookeeper ensemble.

It is important to realize that the client directly handles all communication with the RegionServers, there is no proxy at the server side. Consequently the client needs to do the service discovery and caching as well as the connection and thread management necessary. And hence some of the complexity is understandable: The client is part of the cluster.

See also this blog post. Before HBASE-5682 a client would potentially never recover when it could not reach the cluster. And before HBASE-4805 and HBASE-6326, a client could not - with good conscience - be used in a long running ApplicationServer.

An important aspect of any client library is what I like to call "time to exception". If things go wrong the client should (at least as an option) fail fast and let the calling application - which has the necessary semantic context - decide how to handle this situation.

Unfortunately the HBase and Zookeeper clients were not designed with this in mind.

Among the various time outs are:

• ZK session timeout (zookeeper.session.timeout)
• RPC timeout (hbase.rpc.timeout)
• RecoverableZookeeper retry count and retry wait (zookeeper.recovery.retry, zookeeper.recovery.retry.intervalmill)
• Client retry count and wait (hbase.client.retries.number, hbase.client.pause)

In some error paths these retry loops are nested, so that in the default setting if both ZK and HBase are down a client will throw an Exception after a whooping 20 Minutes! The application has no chance to react to outages in any meaningful way.

HBASE-6326 fixes one issue, where .META. and -ROOT- lookups would be nested, each time causing a ZK timeout N^2 times (N being the client retry count, 10 by default), which itself would be retried by RecoverableZookeeper (3 by default).

The defaults for some of these settings are optimized for the various server side components. If the network "blips" for five seconds the RegionServers should not abort themselves. So a session timeout of 180s makes sense there.

For clients running inside a stateless ApplicationServer the design goals are different. Short timeouts of five seconds seem reasonable. A failure is quickly detected and the application can react (potentially by controlled retrying).

With the fixes in the various jiras mentioned above, it is now possible (in HBase 0.94+) to set the various retry counts and timeouts to low values and get reasonably short timespans after which the client would report a connection error to calling application thread.
And this is in fact what should done when the HBaseClient (HTable, etc) is used inside an ApplicationServer for HBase requests that are synchronous in the calling thread (for example a web server serving data from HBase).

Pluggable compaction and scanning policies

As I described here, HBase maintains multiple versions of all key-values (KVs) stored and has two essential knobs to control the collection old versions: TimeToLive (TTL) and MaximumNumberOfVersion (versions).

These two attributes are "statically" defined for each column family. To change these the table needs to be disabled, the column families changed, and then the table needs to be enabled again. A very heavy weight operation.

An area where this is problematic is M/R based, timestamp consistent, incremental backups (a scheme that I outlined here). Another scenario might be MVCC based transaction engine on top of HBase that use the HBase timestamps to maintain the version.

The problem in both cases is that it is hard to know a priori how long HBase needs to keep older versions around.

In HBASE-6427 I suggest some extensions to the coprocessor framework that would allow a RegionObserver to finely control what KVs are targeted for compaction, by allowing it to create the scanner that is used to scan the incoming KVs.

RegionObserver has three new hooks:

• preFlushScannerOpen - called before a scanner iterating over the MemStore being flushed is created
• preCompactScannerOpen - called before a scanner iterating over all StoreFiles to be compacted is created
• preStoreScannerOpen - called before a user-initiated scan is started

These hooks can return a custom scanner to define the set of KVs that should copied over to the compacted/flushed files.
(The various internal scanner interfaces in HBase are in need of some consolidation work - pre{Flush|Compact}ScannerOpen return an InternalScanner, whereas preStoreScannerOpen return a KeyValueScanner - but that is a different story).
 
HBASE-6427 also makes some HBase internal data structures accessible to coprocessors, so that a simple RegionObserver implementation could return a slightly modified StoreScanner (public API site is not, yet, updated so a link to the Javadoc is not availabel) from any of these hooks.

Looking at test classes introduced by the changed might be interesting, as they have a RegionObserver implement the default logic and then verify that the behavior of various operations is indeed unchanged.

For example now it would be possible for a RegionObserver implementation to listen to a ZooKeeper node, which could indicate what data can safely eliminated (i.e. not be visible to any of the scanner returned from these hooks). An incremental update process can use this to indicate the last time a successful backup was executed.

I intend to provide an implementation of a sample ZK based coprocessor to listen for scan policy changes (HBASE-6496).

Note: Coprocessors are developer tools and not for the casual user. The intend of these changes was to make this form of control possible, not necessarily easy. The very nature of the flush/compaction process makes interfering with them a difficult proposition.

HBase, HDFS and durable sync

HBase and HDFS go hand in hand to provide HBase's durability and consistency guarantees.

One way of looking at this setup is that HDFS handles the distribution and storage of your data whereas HBase handles the distribution of CPU cycles and provides a consistent view of that data.

As described in many other places, HBase

1. Appends all changes to a WAL
2. Batches/sorts changes in memory
3. Flushes memory to immutable, sorted data files in HDFS
4. Combines smaller data files to fewer, larger ones during compactions.

The part that is typical less clearly understood and documented are the exact durability guarantees provided by HDFS (and hence HBase).

HDFS sync has a colorful history, with needed support for HBase only available in an unreleased "append"-branch of HDFS for a long time. (Note that the append and sync features are independent and against common believe HBase only relies on the sync feature). See also this Cloudera blog post.

In order to understand what HDFS provides let's take a look at how a DFSClient (client) interacts with a Datanode (DN).

In a nutshell a DN just waits for commands. One of these commands is WRITE_BLOCK. When the DN receives a WRITE_BLOCK command it instantiates a BlockReceiver thread.

The BlockReceiver then simply waits for packets on an InputStream and flushes the data to OS buffers. An open block is maintained at the DN as an open file. When a block is filled, the block and hence its associate files is closed and the BlockReceiver ends. For all practical purposes the DN forgets that the block existed.

Replication to replica DNs is done via pipelining. The first DN forwards each packet to the next DN in the chain before the data is flushed locally, and waits for the downstreadm DN to respond. The default length of the replication chain is 3.

The other side of the equation is the DFSClient. The DFSClient batches changes until a packet is filled.
Since HADOOP-6313 a Syncable supports hflush and hsync.

• hflush flushes all outstanding data (i.e. the current unfinished packet) from the client into the OS buffers on all DN replicas.
• hsync flushes the data to the DNs like hflush and should also force the data to disk via fsync (or equivalent). But currently for HDFS hsync is implemented as hflush!

The gist is that both closing of a block and issuing hsync/hflush on the client currently only guarantees that data was flushed from the client to the replica DNs and to OS buffers on each DN; not that the data actually reached a physical disk.

For HBase and similar applications with durability guarantees this can be insufficient. If three or more DN machines crash at the same time (assume three replicas), for example caused by a multi rack or data center power outage, data might be lost.
Further, since HBase constantly compacts older, smaller HFiles into newer, larger ones, this potential data loss is not limited to new data.
(But note that, like most database setups, HBase should be deployed with redundant power supply anyway, so this is not necessarily an issue).

Due to the inner working of the DN it is difficult to implement 100% Posix fsync semantics. Imagine a client, which writes many blocks worth of data and then issues an hsync. In order to sync data correctly to disk either the client or all involved DNs would need to keep track of all blocks (full or partial) written to so far that have not yet been sync'ed.

This would be a significant change to how either the client or the DN work and lead to more complicated code. It would also require to keep the block files open in order to retain the file descriptors so that an fsync could be potentially issued in the future. Since the client might in fact never issue a sync request the number of open files to retain is unbounded.

The other option (similar to Posix' O_SYNC) is to have the DNs call fsync upon receipt of every single packet. leading to many unnecessary fsyncs.

In HDFS-744 I propose a hybrid solution. A data stream can be created with a SYNC_BLOCK flag. This flag causes the DFSClient set a "sync" flag on the last packet of a block. I.e. the block file is fsync'ed upon close.

This flag is also set when the client issues hsync. If the client has outstanding data the current packet is tagged with the "sync" flag and sent immediately, otherwise an empty packet with this flags is sent.

When a DN receives such a packet, it will immediately flush the currently open file (representing the current block - full on close or partial on hsync - being written) to disk.

In summary: With this compromise a client can guarantee - byte-by-byte if needed - which portion of an open file is guaranteed on a durable medium while avoiding either syncing every packet to disk or keeping track of past unsync'ed block.
For HBase this would conveniently deal with compactions as blocks are sync'ed upon close and also with WAL edits as it correctly allows sync'ing the current block.

The downside is that upon close each block needs to be sync'ed to disk, even though the client might never issue a sync request for this stream; this leads to potentially unneeded fsyncs.

HBASE-5954 proposes matching changes to HBase to make use of this new HDFS feature. This issue introduces a WAL sync config option and an HFile sync option.
The former causes HBase to issue an hsync when a batch of WAL entries is written. The latter makes sure HFiles (generated from memstore flushes or compactions) are guaranteed to be on a durable medium when the stream is closed.

There are also a few simple performance tests listed in that issue.

Future optimization is possible:

• Only one of the replica DNs could issue the sync
• Only one DN in each Rack could issue the sync
• The sync could be done in parallel and the response from the DN need not wait for for it to finish (in this case the client has no guarantee that the sync actually finished when hsync returns, only that the DN promised to do it)
• For HBase, both options could be made configurable per column family.

HBaseCon

HBaseCon on May 22nd was a blast.

I hosted the development track and presented on HBase Internals (which is essentially a condensed version of this blog). And that all despite my being mortally afraid of public speaking.

The slides of the Internals talk are now available.
Slideshare does not deal well with animation, unfortunately.

I'll add a link to the video once it becomes available.

HBase 0.94

In all the hectic leading up to HBaseCon, I forgot to blog about HBase 0.94.

I released HBase 0.94.0 on May 15, 2012. This was the 4th release candidate. 

0.94 is mostly a performance release with a lot of contributions from Facebook.
As mentioned in the release announcement these are some notable improvements:

• HBASE-5010 - Filter HFiles based on TTL
• HBASE-4465 - Lazy-seek optimization for StoreFile scanners
• HBASE-4469 - Avoid top row seek by looking up ROWCOL bloomfilter
• HBASE-4532 - Avoid top row seek by dedicated bloom filter for delete family bloom filter
• HBASE-5074 - support checksums in HBase block cache

among others. Many thanks to Mikhail Bautin and others at Facebook for all these improvements.

Of course there are some new features as well.

• Jonathan Hsieh's HBASE-5128 - [uber hbck] Online automated repair of table integrity and region consistency problems
• Jon Gray's HBASE-4460 - Support running an embedded ThriftServer within a RegionServer

And then of course my pets (I blogged about these before):

• HBASE-3584 - Allow atomic put/delete in one call
• HBASE-4102 - atomicAppend: A put that appends to the latest version of a cell
• HBASE-4536 - Allow CF to retain deleted rows
• HBASE-5229 - Provide basic building blocks for "multi-row" local transactions.

0.94 is binary compatible with 0.92. Server and client can be freely mixed between versions, it is even possible to mix servers of different version within the same cluster (for rolling upgrades).

I expect 0.94.x to be used for a while until the "singularity" (0.96) stabilizes into HBase 1.0.

Update Wednesday, May 30th, 2012:
HBase can be downloaded from here: http://www.apache.org/dyn/closer.cgi/hbase/
The list of features and bug fixes is here.

(timestamp-) Consistent backups in HBase

The topic consistent backups in HBase comes up every now and then.
In this article I will outline a scheme that does provide timestamp-consistent backups.

Consistent backups are possible in HBase. With "consistent" I mean "consistent as of a specific timestamp" (this is limited by the HBase timestamp granularity.)

Over the past few months I have contributed various changes to HBase that now hopefully lead full circle to a more coherent story for "system of record" type data retention.

The basic setup is simple. You'll need

• HBASE-4071 (0.92 - for MIN_VERSIONS)
• HBASE-4536 (0.94 - keeping deleted cells and raw scans)
• HBASE-4682 (0.94 - deleted cells in exports)

(HBase 0.94 will have all the relevant patches once it is releases - which should be any time now)

HBase's built-in Export tool can then be used to generate consistent snapshots.

Normally the data collected by an Export job is "smeared" over the time interval it takes to execute the job; an Export-Scan sees a row (cells of a row to be precise) as of the time when it happens to get to it, and rows can change while the Export is running.

That is problematic, because it is not possible to recreate a consistent view of the database from export generated this way.

The trick then is to keep all data in HBase long enough for the backup job to finish, and to only collect information before the start time of the export job.

Say the backup takes no more than T to complete (yes, it is hard to know ahead of time how long an Export is going to run). In that case the table's column families can be setup as follows:

• set TTL to T + some headroom (so maybe 2T to be safe)
• set VERSIONS to a very large number, (max int = 2147483647 for example, i.e. nothing is evicted from HBase due to VERSIONS)
• set MIN_VERSIONS to how many versions you want to keep around, otherwise all versions could be removed if their TTL expired
• set KEEP_DELETED_CELLS to true (this makes sure that deleted cell and delete markers are kept until they expire by the TTL)

In the shell (setting TTL to two hours):

create <table>, {NAME=><columnfamily>, VERSIONS=>2147483647, TTL=>7200, MIN_VERSIONS=>1, KEEP_DELETED_CELLS=>true}

Export can now we used as follows:

• set versions to 2147483647 (i.e. all versions)
• set startTime to -2147483648 (min int, i.e. everything is guaranteed to be included)
• set endTime to the start time of the Export (i.e. the current time when the export starts. This is the crucial part)
• enable support for deleted cell

hbase org.apache.hadoop.hbase.mapreduce.Export
-D hbase.mapreduce.include.deleted.rows=true
<tablename> <outputdir>
2147483647 -2147483648 <now in ms>

As long as the Export finishes within 2T, a consistent snapshot as of the time the Export was started is created. Otherwise some data might be missing, as it could have been compacted away before the Export had a chance to see it.

Since the backups also copied deleted rows and delete markers, a backup restored to an HBase instance can be queried using a time range (see Scan) to retrieve the state of the data at any arbitrary time.

Export is current limited to a single table, but given enough storage in your live cluster this can be extended to multiple table Exports, simply by setting the endTime of all Exports jobs to the start time of the first job.


This same trick can also be used for incremental backups. In that case the TTL has to be large enough to cover the interval between incremental backups.
If, for example, the incremental backups frequency is daily, the TTL above can be set to 2 days (TTL=>172800). Then use Export again:

hbase org.apache.hadoop.hbase.mapreduce.Export
-D hbase.mapreduce.include.deleted.rows=true
<tablename> <outputdir>
2147483647 <time of last backup in ms> <now in ms>

The longer TTL guarantees that there will be no gaps that are not covered by the incremental backups.

An example:

1. A Put (p1) happens at T1
2. Full backup starts at T2, time interval [0, T2)
3. Another Put (p2) at T3
4. full backup jobs finishes
5. A Delete happens at T4 
6. Incremental backup starts at T5, time interval [T2, T5)
7. Yet another Put (p3)
8. Incremental backup finishes

Note that in this scenario is does not matter when the backup jobs finish.

The full backup contains only p1. The incremental backup contains p2 and the Delete. p3 is not included in any backup, yet.
The state at T2 (p1) and T5 (p1, p2, delete) can be directly restored. Using time range Scans or Gets the state as of T4 and T3 can also be retrieved, once both backups have been restored into the same HBase instance (you need HBASE-4536 for this to work correctly with Deletes).

Finally, if keeping enough data to cover the time between two incremental backups in the live HBase cluster is problematic for your organization, it is also possible to archive HBase's Write Ahead Logs (WAL) and then replay with the built-in WALPlayer (HBASE-5604), but that is for another post.