Clustrix

Why Build an Appliance?

2012-02-01T12:28:00.000-08:00

Clustrix sells appliances. We marry our software with industry standard hardware to make plug and play devices that drop in and work on the network. This is same model as my former company Isilon (now EMC), NetApp, and pretty much all successful storage vendors. Why do this instead of selling software by itself?

Qualifying SSDs is particularly tricky because of the huge variation in quality.

The number one reason for the appliance model is quality control. By reducing the supported hardware set, we drastically increase the QA time we get on the intended hardware. QA time looking for hardware interactions really matter in a product that stores customers’ data. The bar is simply set much higher there. As an example, we do extensive testing of durability of data on our specific hardware. Depending on the hard drive controller, controller firmware, drives, and drive firmware, we have gotten very different results on exactly when a specific piece of data makes it to stable storage. Many drives and even controllers lie about when the data is safe. Some drives implement tagged queuing or FUA poorly. I’ve even seen drives return from a sync command without actually having synced the data. The only way to ensure data integrity on a storage system is to properly characterize the hardware with an extensive test regime and control every piece of that storage system. With that control, we can form relationships with the vendors to fix bugs that we expose in the hardware and firmware. At Clustrix, we have a variety of different pieces of test software to exercise the disk subsystem and verify the data is safe every time. With that test software, we have rejected many pieces of hardware and countless versions of firmware that didn’t make the cut. The same rigorous process goes into qualifying networking, Infiniband, NVRAM, processing, and memory components. This kind of focused testing and qualification is not possible on a software-only product.

The second critical benefit of the appliance model is much tighter integration between hardware and software. At Clustrix, we are able to monitor the hardware in the box and present that data seamlessly in our “system” database. For example, “SELECT * from system.memory” will give you all the details for the memory installed on the system and tell you if there are any correctable or uncorrectable ECC errors. We have logic to send alerts on correctable errors and safely shut down the node on an uncorrectable error. On Clustrix, this hardware-specific data sits right along side of database-specific data like queries per second and disk full percentage which allows exceptionally easy integration with tools like Nagios and Cacti. This sort of tight integration is only possible on an appliance.

Finally, being an appliance makes the Clustrix database much easier to install, manage, and use. Creating a high performance, bullet-proof database is no longer a science project. You no longer have to put together the pieces, get all the right versions of firmware, get the right versions of the kernel and libraries, the right version of the database software and make sure all the parts are tuned to work together. The Clustrix appliance is an integrated and tuned database right out of the box.

Scripting SSH with Python

2012-01-31T15:04:00.000-08:00

The other day Aaron Passey, our CTO, pointed out to me that I've written the same bit of code in different languages at each of the last 3 places I've worked. I figured that since this seems to be something commonly useful, yet not obviously available, I'd write something about it.

The scenario is that we have some code which wants to do remote ssh calls. Some variant of this code exists within the Isilon cluster management code, and we use it at Clustrix within our clx command line tool, as well as the database update scripts. We're already running on a Unix box (or variant (osx, linux, etc)) and we have access to an ssh client, so this it totally do-able from code, but not immediately obvious how.

I'll present this in Python, but the same applies to C, or any other language. I'm aware of the Paramiko library for Python which is supposed to have support for this. That may do everything this can do - I don't know. This is 100 lines of code, pretty easy to follow, and maybe instructive. This is all about utilizing ssh and our friendly posix primitives. After the walkthrough, I've included the entire source at the bottom.

The obvious thing that we'd like to do is fork off an ssh process and read the stdin. The easy way to do this in python is with os.popen2(). This will give us back the stdin and stdout:

(sin, sout) = os.popen2(cmd)

This, however, will not work. ssh wants a psuedo tty (a pty). If it's not running in one, it just exits. This is where the helpful python pty class comes in:

(pid, f) = pty.fork()

Now we've got ssh running in the right environment. The two args we got back are important: pid is the process id of our forked ssh process, and f is a unix fileno (not to be confused with a python file handle) which is the combined stdin and stdout of the process. It's important to remember that f isn't something reference counted. We're going to need to explicitly close it.

Now that we have the basic mechanism in place, let's build us a little ssh class:

class SSH:
    def __init__(self, ip, passwd, user, port):
        self.ip = ip
        self.passwd = passwd
        self.user = user
        self.port = port

This is structured so you can create one SSH object per target box and reuse it to do different commands. We'll also include the ability to push and pull files. Our first method will be the command handler. It will take only one argument (other than self), which will be the command to run:

    def run_cmd(self, c):
        (pid, f) = pty.fork()
        if pid == 0:
            os.execlp("ssh", "ssh", '-p %d' % self.port,
                      self.user + '@' + self.ip, c)
        else:
            return (pid, f)

As you can see, the pty fork works just like the os fork in that if the pid is 0 it means we're the child, and if non 0 it means we're the parent.

Since this is a raw unix fileno, the file closed condition is a little weird. Reading it will block until something is available and then return results. But when the descriptor closes it throws an os error. I'd rather be able to handle the reads just in a loop (or maybe it's because I'm an old C programmer) so I'm going to wrap the read:

    def _read(self, f):
        x = ''
        try:
            x = os.read(f, 1024)
        except Exception, e:
            # this always fails with io error
            pass
        return x

Once we've got this thing forked, and can read from it, we need to get our results back out. There's on additional thing we need to be prepared for: ssh might want to ask us some questions. We've all seen this before:

harmony:~$ ssh paulmini
The authenticity of host 'paulmini (10.1.2.125)' can't be established.
RSA key fingerprint is 7e:91:5d:5d:06:fe:3f:24:94:84:

c0:75:96:8c:d1:f1.
Are you sure you want to continue connecting (yes/no)?

We just want to say "yes" and move on. ssh might also ask us for a password if we don't have host keys enabled and we'll need to be prepared to handle that. We'll handle all of these actions in an ssh_results method:

    def ssh_results(self, pid, f):

First, let's initialize our output buffer:

        output = ""

Now we'll read our first chunk and see if ssh is asking anything of us. If they want to know if we really want to continue connecting because the target isn't in ssh/known_hosts, we'll say yes. If they ask us for the password we'll provide it.

        got = self._read(f)

        # check for authenticity of host request
        m = re.search("authenticity of host", got)
        if m:
            os.write(f, 'yes\n')
            # Read until we get ack
            while True:
                got = self._read(f)
                m = re.search("Permanently added", got)
                if m:
                    break

            got = self._read(f)         # check for passwd request
        m = re.search("assword:", got)
        if m:
            # send passwd
            os.write(f, self.passwd + '\n')
            # read two lines
            tmp = self._read(f)
            tmp += self._read(f)
            m = re.search("Permission denied", tmp)
            if m:
                raise Exception("Invalid passwd")
            # passwd was accepted
            got = tmp

Preliminaries done, we can now entire our results loop:

        while got and len(got) > 0:
            output += got
            got = self._read(f)

We've know we've now ready everything because our _read returned empty. This also means that the ssh process has ended. There's a defunct zombie process sitting there and we're going to have to clean that up. We could handle the SIGCHLD and waitpid it, but in this case since we know the pid and we know the process is done, it's much simpler. Also, remember that since f isn't a reference counted Python object we're going to need to manually clean that up:

        os.waitpid(pid, 0)
        os.close(f)
        return output

And that's basically it. With some niceties for pushing and pulling files, and some error handling, the completed code looks like this:

#
# Remote ssh cmds
#

import pty, re, os, sys, stat, getpass

class SSHError(Exception):
    def __init__(self, value):
        self.value = value
    def __str__(self):
        return repr(self.value)

class SSH:
    def __init__(self, ip, passwd, user, port):
        self.ip = ip
        self.passwd = passwd
        self.user = user
        self.port = port

    def run_cmd(self, c):
        (pid, f) = pty.fork()
        if pid == 0:
            os.execlp("ssh", "ssh", '-p %d' % self.port,
                      self.user + '@' + self.ip, c)
        else:
            return (pid, f)     def push_file(self, src, dst):
        (pid, f) = pty.fork()
        if pid == 0:
            os.execlp("scp", "scp", '-P %d' % self.port,
                      src, self.user + '@' + self.ip + ':' + dst)
        else:
            return (pid, f)

    def push_dir(self, src, dst):
        (pid, f) = pty.fork()
        if pid == 0:
            os.execlp("scp", "scp", '-P %d' % self.port, "-r", src,
                      self.user + '@' + self.ip + ':' + dst)
        else:
            return (pid, f)

    def _read(self, f):
        x = ''
        try:
            x = os.read(f, 1024)
        except Exception, e:
            # this always fails with io error
            pass
        return x

    def ssh_results(self, pid, f):
        output = ""
        got = self._read(f)         # check for authenticity of host request
        m = re.search("authenticity of host", got)
        if m:
            os.write(f, 'yes\n')
            # Read until we get ack
            while True:
                got = self._read(f)
                m = re.search("Permanently added", got)
                if m:
                    break

            got = self._read(f)         # check for passwd request
        m = re.search("assword:", got)
        if m:
            # send passwd
            os.write(f, self.passwd + '\n')
            # read two lines
            tmp = self._read(f)
            tmp += self._read(f)
            m = re.search("Permission denied", tmp)
            if m:
                raise Exception("Invalid passwd")
            # passwd was accepted
            got = tmp
        while got and len(got) > 0:
            output += got
            got = self._read(f)
        os.waitpid(pid, 0)
        os.close(f)
        return output

    def cmd(self, c):
        (pid, f) = self.run_cmd(c)
        return self.ssh_results(pid, f)

    def push(self, src, dst):
        s = os.stat(src)
        if stat.S_ISDIR(s[stat.ST_MODE]):
            (pid, f) = self.push_dir(src, dst)
        else:
            (pid, f) = self.push_file(src, dst)
        return self.ssh_results(pid, f)

def ssh_cmd(ip, passwd, cmd, user=getpass.getuser(), port=22):
    s = SSH(ip, passwd, user, port)
    return s.cmd(cmd)

def ssh_push(ip, passwd, src, dst, user=getpass.getuser(), port=22):
    s = SSH(ip, passwd, user, port)
    return s.push(src, dst)

Running a devnode cluster across multiple boxes

2012-01-27T11:33:00.000-08:00

In my last post, we used devnodectl to fire up a simple 3-node cluster, with all nodes running on the same system. This is interesting insofar as it demonstrates the functionality of the cluster, but we're certainly quite resource limited trying to emulate three nodes on one box. In this post, I'll show how to get devnode running on different physical servers, so you can begin to see the potential of horizontal scaling available with Clustrix.

Along the way I'll also cover:

Manual control of devnode (vs. devnodectl)
How, where, and what to look for in the logs

Requirements

Three Red Hat/CentOS 6 or equivalent clients
All clients must be on the same subnet
The clients should NOT be running mysqld (so port 3306 will be free to devnode)

Firing up the devnode instances

First install the DevKit RPMs, as described in my last blog post, on each of your clients, and ensure you have a writeable working directory available on each (I'll use /data/clustrix on each). If you followed along on the last exercise, please stop those nodes (devnodectl stop), and I also recommend cleaning out the prior state with rm -rf /data/clustrix/*. The flags we'll specify below will overwrite old stuff as needed, but you'd still have the old nodes' 2 and 3 state on your first client, which might become confusing.

Now we're going to start up devnode directly, and also change the flags around quite a bit. I recommend opening three terminal windows, one for each node; connect (ssh, presumably) to each of your clients with these nodes, then run the following, one in each window:

client1$ /opt/clustrix/bin/devnode -clusterpath /data/clustrix -setpnid 1 -eth eth0 -clean -nclean 4 -vdev-size 2048 -logfile -

client2$ /opt/clustrix/bin/devnode -clusterpath /data/clustrix -setpnid 2 -eth eth0 -clean -nclean 4 -vdev-size 2048 -logfile - -noautostart

client3$ /opt/clustrix/bin/devnode -clusterpath /data/clustrix -setpnid 3 -eth eth0 -clean -nclean 4 -vdev-size 2048 -logfile - -noautostart

The -clusterpath, -nclean, and -vdev-size flags we talked about last time (recap: where does the simulated node store it's data, and how many/big disks should we have -- note that I'm allocating 8GB per node here). Let's pick apart the other flags here:

-setpnid sets a Physical Node ID -- normally this would come from a node's MAC address
-eth tells devnode to use the ethernet interface for inter-node communication; when we created our three node cluster on the same physical machine with devnodectl, it specified -unix to use UNIX sockets instead. On real nodes, we'd be using InfiniBand for this purpose.
-clean tells the cluster to wipe all prior state
-noautostart for the second and third nodes avoids a devnode restart step, as will be explained further below
-logfile - means to log to stdout instead of to a log file

This last option is how I prefer to run, because staring at logfiles is how I live. For our purposes today I think it will be most instructive for you as well. It should be noted that normally these logs go to /data/clustrix/p1/devnode.log (substitute p2, p3, etc. for other nodes).

If a bunch of FATAL errors scroll by, the most likely culprit is a port conflict:

2012-01-25 11:22:38 ERROR cp/cp_sock.c:74 cp_bind(): stream_listen(IPv4(0.0.0.0:2048)): Address already in use

2012-01-25 11:22:38 FATAL core/segv.c:93 main_segv_handler(): Program received a fatal signal on core 0 fiber 0

2012-01-25 11:22:38 FATAL core/segv.c:95 main_segv_handler(): C stack trace:

0x000000000052d976 bind_done() <no lines read>

0x00000000008c7c91 scheduler_run_one_item() <no lines read>

0x00000000008c8893 scheduler_main_loop() <no lines read>

The above indicates that port 2048 (the control port, which will cover later) is already in use. You'd see this if you tried to run the above commands on the same box, instead of 3 different boxes. If you're running mysqld on one of your boxes, it will fail thusly:

n1 2012-01-25 11:37:47 ERROR mysql/server/mysql_proto.c:1171 listen_on_port(): stream_listen(IPv4(0.0.0.0:3306)): Address already in use

n1 2012-01-25 11:37:47 FATAL dbcore/dbstate.c:104 dbconf_done(): Error handling dbconf chain: Address already in use: dbconf/INIT_MYSQL_PROTO failed (unable to create TCP socket)

You can work around these by specifying the -anyport flag, in which case you'll need to look back through the logs to find which port it's chosen:

2012-01-25 11:41:17 INFO dbcore/driver.ct:90 driver_publish_address(): pnid p3 control port 33274 sw f71a74d89c512ac

n1 2012-01-25 11:41:17 INFO dbcore/driver.ct:90 driver_publish_address(): pnid p3 mysql port 36290 sw f71a74d89c512ac

Adding nodes 2 and 3 to your cluster

Normally fresh Clustrix nodes start up in a "cluster-of-one"; you can connect to any node and then pull in the others to form a larger cluster. This mechanism involves a process restart (on real nodes, an initd-like process called nanny takes care of this); to avoid this, we used the -noautostart flag when starting nodes 2 and 3, so they don't start up as "cluster-of-one", and can't be accessed via mysql until they are added to a cluster.

So, connect to your first client (client1 above, the one where you did not have the -noautostart flag):

[nparrish@hefty mainline1]$ mysql -h beta001 -u root

mysql> use system;

Reading table information for completion of table and column names

You can turn off this feature to get a quicker startup with -A

Database changed

mysql> select * from available_node_details;

+------+------------+----------------------------------------------------+

| pnid | name | value |

+------+------------+----------------------------------------------------+

| p3 | cluster | nparrish |

| p3 | sw_version | 1112854534402937516 |

| p3 | version | 5.0.45-clustrix-v3.2-7371-0f71a74d89c512ac-release |

| p3 | iface_ip | 10.2.12.188 |

| p3 | iface_mac | 00:25:90:34:69:04 |

| p3 | hostname | loeb.colo.sproutsys.com |

| p3 | started | 2012-01-25 20:44:33.037318 |

| p2 | cluster | nparrish |

| p2 | sw_version | 1112854534402937516 |

| p2 | version | 5.0.45-clustrix-v3.2-7371-0f71a74d89c512ac-release |

| p2 | iface_ip | 10.2.12.194 |

| p2 | iface_mac | 00:25:90:34:70:0a |

| p2 | hostname | sainz.colo.sproutsys.com |

| p2 | started | 2012-01-24 23:52:17.085502 |

+------+------------+----------------------------------------------------+

14 rows in set (0.00 sec)

So we're looking at system.available_node_details, which shows the nodes that can be seen on the network (here eth0, on real nodes via InfiniBand), who are not already part of another cluster. It's sort of a key/value pair table, extended to include the pnid (Physical Node ID -- recall we set this with -setpnid). We're really most interested in the hostname -- I can see my two other clients, great.

Now we add these nodes into the cluster using the ALTER CLUSTER query:

mysql> alter cluster add p2;
Query OK, 0 rows affected (0.00 sec)

mysql> alter cluster add p3;
Query OK, 0 rows affected (0.02 sec)

mysql> select * from nodeinfo\G
*************************** 1. row ***************************
nodeid: 1
started: 2012-01-25 20:50:50
ntptime: 2012-01-25 20:54:13
node uptime: 2012-01-20 23:15:52
hostname: beta001.colo.sproutsys.com
iface_name: eth0
iface_ip: 10.2.13.11
iface_mac_addr: 00:30:48:c3:e7:5c
pnid: p1
cores: 12288
*************************** 2. row ***************************
nodeid: 3
started: 2012-01-25 20:50:57
ntptime: 2012-01-25 20:54:13
node uptime: 2011-04-12 01:05:08
hostname: loeb.colo.sproutsys.com
iface_name: eth0
iface_ip: 10.2.12.188
iface_mac_addr: 00:25:90:34:69:04
pnid: p3
cores: 805313044
*************************** 3. row ***************************
nodeid: 2
started: 2012-01-25 20:50:53
ntptime: 2012-01-25 20:54:13
node uptime: 2011-10-25 17:31:38
hostname: sainz.colo.sproutsys.com
iface_name: eth0
iface_ip: 10.2.12.194
iface_mac_addr: 00:25:90:34:70:0a
pnid: p2
cores: 3145744
3 rows in set (0.01 sec)

And you're ready to rock and roll. (Yes, the cores value is a little funny -- on Clustrix nodes this would tell you how many CPU cores available on each node.)

Accessing Your Cluster

With each node using the standard MySQL port, it's a little less fussy to connect to the cluster, as you no longer need to find and specify a different port for each node. As before, connect to any node and you see the same database instance.

Normally a Clustrix cluster is configured with a Virtual IP (VIP), which is a distinct IP address which load balances connections across all the nodes. This is partially implemented within the base OS of our appliance nodes, so unfortunately we cannot provide this functionality with the DevKit. You can, however, implement simple load-balancing with a tool like HAProxy (we use this internally as a simple solution to cut over between clusters located on different subnets, where DSR is not possible). I'll see about writing this up as another blog post.

Please bear in mind that while we've now got our devnode processes distributed across multiple servers, they are now communicating over ethernet instead of via UNIX sockets (memory). Ethernet is a far cry from the InfiniBand which connects real Clustrix nodes. Beyond IB being a higher throughput, lower latency interconnect, our software is tuned for these characteristics, so we're not going to see world-beating performance with our simulated cluster running over ethernet. For that, I'll refer you to our prior blog post on Percona's TPCC test!

Recap

So we've shown how to get devnode running on different nodes, communicating with each other via ethernet. While a little more "real" than having them all running on a single box, the performance characteristics are going to be orders of magnitude off from the capabilities of proper Clustrix nodes. This does provide a simulacra of the platform to develop against, and we'd welcome the opportunity to move you from there to deploying on real hardware. As always, your feedback or questions are welcome, in the comments or support forum.

Getting started with Clustrix DevKit

2012-01-23T17:23:00.000-08:00

We recently released our Clustrix Developers Kit to enable folks to experiment with the Clustrix solution right away, without having to take delivery of actual hardware or coordinate a hosted evaluation. The kit includes the devnode binary, which is the database code minus the few hardware-specific bits to run on our appliance hardware, and a devnodectl utility to simplify wrangling multiple instances of devnode together into a functional cluster. With this kit, it is possible to create a fully functional cluster, load data onto it, and run your application against it. Obviously there are limitations: CPU, memory, and disk I/O capacity are going to be divided between multiple simulated nodes, and while we can simulate failure scenarios (as we will do in a future post), actual fault tolerance requires Clustrix nodes.

In this post, I'll walk you through:

Installing the RPMs that comprise the DevKit
Using devnodectl to start up a cluster
Accessing the database instance with the mysql client
Dumping and importing data from a MySQL instance
Starting MySQL replication as a slave

Along the way I want to highlight the following features of Clustrix:

Drop-in compatibility with MySQL, including replication
Single-instance database (no sharding!)

Let's get started!

Installation

First download from the main page at www.clustrix.com, the "Try it now" button. You'll need to download and install the common and devnode packages as follows:

[root@beta001 v3.2]# rpm -ivh clustrix-common-v3.2-493.x86_64.rpm clustrix-devnode-v3.2-7371.x86_64.rpm

Preparing... ########################################### [100%]

1:clustrix-common ########################################### [ 50%]

2:clustrix-devnode ########################################### [100%]

If you run into any errors installing the RPMs, it's likely to be due to missing dependencies, which you may be able to fill in using yum. Otherwise, you'll probably need to find a client with a more recent Redhat/CentOS install (for this exercise, I'm using CentOS 6).

What does the RPM actually install? Everything installs under /opt/clustrix; support libraries in

lib/, and the two executables we care about in bin/: devnode and devnodectl.

You'll also need a data directory for the cluster to store files which will serve as virtual disks. It is highly recommended that you store these on local disk rather than over NFS. Bear in mind that with proper Clustrix nodes, this storage is provided with SSDs, thus the throughput capacity of your local storage will be a constraining factor in performance of a devnode cluster (a subject I plan to revisit in a future blog post). Here I'm going to use /data/clustrix, which is the default for the devnodectl utility (if you use a different dir, you'll need to use the -d/--data-dir argument each time you run devnodectl):

[root@beta001 v3.2]# mkdir /data/clustrix

[root@beta001 v3.2]# chown nparrish /data/clustrix

If you are running as a non-root user (which is advised), you'll also want to chown the directory to your username, as shown.

Initializing the cluster

devnodectl takes care of starting up devnode processes, specifying flags so that they automatically join together into a cluster. In later blog posts we'll take a more manual approach, in order to demonstrate things like adding nodes to an existing cluster, running devnode instances on multiple clients, and fault tolerance features. But for now, firing up the cluster is as simple as:

[nparrish@beta001 ~]$ /opt/clustrix/bin/devnodectl --init --nodes 3 start

Let's look at the output this returns:

[exec] /opt/clustrix/bin/devnode -clusterpath /data/clustrix -cluster nparrish -setpnid 1 -anyport -unix -glue -noautostart -nclean 4 -vdev-size 256

[exec] /opt/clustrix/bin/devnode -clusterpath /data/clustrix -cluster nparrish -setpnid 2 -anyport -unix -noautostart -nclean 4 -vdev-size 256

[exec] /opt/clustrix/bin/devnode -clusterpath /data/clustrix -cluster nparrish -setpnid 3 -anyport -unix -noautostart -nclean 4 -vdev-size 256

This shows us the actual devnode commands being run. I'll draw your attention to a few of the options specified:

-clusterpath /data/clustrix: All cluster state, including the files used for virtual disks, will be in this dir.

-nclean 4 -vdev-size 256: This gives us 4 disks of 256MB each, for each of the three nodes.

Waiting for node 1 to enter quorum.

done.

Node 1 [RUNNING]: /data/clustrix/p1

mysql: 3306 control: 52924 healthmon:3581

mysql socket: /data/clustrix/p1/mysql.sock

Node 2 [RUNNING]: /data/clustrix/p2

mysql: 57059 control: 57079 healthmon:52498

mysql socket: /data/clustrix/p2/mysql.sock

Node 3 [RUNNING]: /data/clustrix/p3

mysql: 58157 control: 2048 healthmon:52002

mysql socket: /data/clustrix/p3/mysql.sock

Example command to access your cluster:

mysql -u root -S /data/clustrix/p3/mysql.sock

This tells us how to access each of the three nodes that have been started. The -anyport command means that each node gets a random MySQL port (after trying for the default 3306, which you can see node 1 gets -- if your client is already running a mysqld instance, all nodes would get some different port number), and this output tells you what port to specify for your clients to connect. You can also connect via socket, as shown in the example. The control and healthmon ports will be covered in a future blog post.

Connecting to the cluster

[nparrish@beta001 ~]$ mysql -h 127.0.0.1 -P 58157 -u root

Welcome to the MySQL monitor. Commands end with ; or \g.

Your MySQL connection id is 3074

Server version: 5.0.45

This software comes with ABSOLUTELY NO WARRANTY. This is free software,

and you are welcome to modify and redistribute it under the GPL v2 license

Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.

mysql> select @@version;

+----------------------------------------------------+

| @@version |

+----------------------------------------------------+

| 5.0.45-clustrix-v3.2-7371-0f71a74d89c512ac-release |

+----------------------------------------------------+

1 row in set (0.00 sec)

A few things to note here:

This is standard, off-the-shelf mysql command line client (hence the Oracle copyright)
I connected using the IP for localhost; mysql client tries to be smart and use the default mysql socket (/var/lib/mysql/mysql.sock) if you specify -h localhost
Clustrix reports server version 5.0.45 for compatibility only; from the full version string in @@version, the relevant Clustrix part starts at v3.2)
We are connecting with built-in root user, which initially has no password

Let's create a table and insert a little bit of data:

mysql> use test;

Database changed

mysql> create table foo (id int key, v varchar(100));

Query OK, 0 rows affected (0.09 sec)

mysql> insert into foo values (1, 'bar'), (2, 'baz');

Query OK, 2 rows affected (0.00 sec)

mysql> select * from foo;

+----+------+

| id | v |

+----+------+

| 1 | bar |

| 2 | baz |

+----+------+

2 rows in set (0.03 sec)

Nothing fancy, but let's now connect to a different node and see how things look:

mysql> exit

Bye

[nparrish@beta001 ~]$ mysql -h 127.0.0.1 -P 57059 -u root test

Reading table information for completion of table and column names

You can turn off this feature to get a quicker startup with -A

[...]

mysql> show tables;

+----------------+

| Tables_in_test |

+----------------+

| foo |

+----------------+

1 row in set (0.00 sec)

mysql> select * from foo;

+----+------+

| id | v |

+----+------+

| 2 | baz |

| 1 | bar |

+----+------+

2 rows in set (0.01 sec)

Note that here I've connected to a different node, but I have the exact same view of my data. You might also note that the rows were returned in a different order. On Clustrix, rows are distributed across multiple nodes, and so if no ORDER BY clause is specified, may be returned in different order. This differs from MySQL, which implicitly orders by primary key (since it must only read from one place).

mysqldump and clustrix_import

Let's import data from an existing MySQL instance (here running on a different client, called hefty). We start by dumping with mysqldump:

[nparrish@hefty tmp]$ mysqldump -h 127.0.0.1 --single-transaction --master-data=2 sbtest > /tmp/sbtest.sql

--single-transaction ensures that all tables in the database are consistent with respect to eachother
--master-data=2 gives us the position in the replication binlog that corresponds to that single transaction, so we can start up a replication slave
Here we are just dumping a specific database, sbtest. We'll also need to create this database on our cluster, as the resulting dump file will not have DROP/CREATE DATABASE statements for it.

We can then import this using mysql client:

[nparrish@hefty tmp]$ mysql -h beta001 -P 3306 sbtest < /tmp/sbtest.sql

A few things to note here:

As noted, we need to indicate which database to import into, so we must first CREATE DATABASE sbtest; then specify sbtest argument to indicate the target database.
If we were importing into actual nodes, we'd use a tool called clustrix_import, which does the inserts in parallel; the tool has not been (de)tuned for use with devnode, so using these together will result in an out-of-memory condition.

Recall that these nodes were created with 4 256MB disks each, so by default the cluster will accomodate only a small (~1GB) dump file. If you want larger disks, use the --drive-size option when running devnodectl --init (see devnodectl -h output for details).

MySQL Replication

We'll dedicate a whole future blog post to replication, but for just a taste let's start up a slave to catch up from the point at which we took the above dump. First we need to find the log file and position, which is stored as a comment in our dump file:

[nparrish@hefty tmp]$ head -24 /tmp/sbtest.sql

-- MySQL dump 10.11

[....]

-- Position to start replication or point-in-time recovery from

-- CHANGE MASTER TO MASTER_LOG_FILE='hefty-bin.000001', MASTER_LOG_POS=610868;

So we just need to add the hostname and user account to use in order to point our cluster at this MySQL instance (note that we can run this on any of our nodes):

mysql> CHANGE MASTER TO MASTER_LOG_FILE='hefty-bin.000001', MASTER_LOG_POS=610868, MASTER_HOST='hefty', MASTER_USER='root';

Query OK, 0 rows affected (0.04 sec)

mysql> start slave;

Query OK, 0 rows affected (0.02 sec)

mysql> show slave status\G

*************************** 1. row ***************************

Slave_Name: default

Slave_Status: Running

Master_Host: hefty

Master_Port: 3306

Master_User: root

Master_Log_File: hefty-bin

Slave_Enabled: Enabled

Log_File_Seq: 2

Log_File_Pos: 267

Last_Error: no error

Connection_Status: Connected

Relay_Log_Bytes_Read: 0

Relay_Log_Current_Size: 0

Seconds_Behind_Master: 0

1 row in set (0.01 sec)

And we're off and replicating any write events on the MySQL instance.

Stopping devnodes

To shut down your cluster, run devnodectl stop:

[nparrish@beta001 clustrix]$ /opt/clustrix/bin/devnodectl stop

[node /data/clustrix/p1] kill -9 21089

[node /data/clustrix/p2] kill -9 20078

[node /data/clustrix/p3] kill -9 20079

As noted, the processes are simply stopped with kill -9. You can restart them with devnodectl start, this time leaving out the --init and --nodes options; note that these processes will probably get different mysql ports than for the prior run. We'll spend more time with restarting devnode processes when we cover fault tolerance in a future post.

Recap

So we've covered some of the initial steps you'll be taking with the Clustrix developers kit:

Installing the RPMs
Starting up a simple three node cluster with devnodectl
Connecting to the cluster with mysql
Using mysqldump to get data from an existing MySQL instance, then import with mysql onto your cluster
Using MySQL replication to set up your cluster as a slave to that MySQL instance

If you're familiar with MySQL, most of this should look pretty familiar, and that's quite the point -- the drop-in compatibility you get with Clustrix means there's no application rewrite involved. Out of the box you get a cluster which hosts a single-instance database, no federation or sharding mess.

What's next

Stay tuned for more. Future topics will include a closer look at how data is stored on Clustrix, deeper coverage of replication capabilities, running devnode on multiple hosts, fault tolerance, examining query performance, and more.

Please come talk to us on the support forums at https://groups.google.com/a/clustrix.com/group/support-public/topics if you run into any problems or have more technical questions.

Distributed Database Architectures: Distributed Storage / Centralized Compute

2011-11-04T18:25:00.000-07:00

In my previous post I wrote about shared disk architectures and the problems they introduce. It's common to see comparisons between shared disk and shared nothing architectures, but that distinction is too coarse to capture the differences between various shared nothing approaches.

Instead, I'm going to characterize the various "shared-nothing" style systems by their query evaluation architectures. Most systems fall into one of the following buckets:

Centralized compute
Limited distributed compute
Fully distributed compute

Centralized Compute: MySQL Cluster

MySQL Cluster consists of two basic roles used for servicing user queries: a compute role and a storage/data role. The compute node is the front end which takes in the query, plans it, and executes it. The compute node will communicate with the storage nodes remotely to fetch any data relevant to the query.

In the distributed storage model, data is no longer shared between the nodes at the page level. Instead, the storage nodes expose a higher level API which allows the compute node to fetch row ranges based on the available access paths (i.e. indexes).

In such a system, storage level locks associated with the data are now managed exclusively by the storage node itself. A compute node does not cache any data; instead, it always asks the set of storage nodes responsible for the data. The system solved the cache coherence overhead problem.
However, it still suffers from extensive data movement and centralized query evaluation.

~~Cache coherence overhead~~
Extensive data movement
Centralized query evaluation

MySQL Query Evaluation in Action

Consider the following example:

SELECT count(*) FROM mytable WHERE acol = 1 and bcol = 2

Assumptions:

an index over acol
10% of the rows in the table match acol = 1
3% of the rows match acol = 1 and bcol = 2
total table size 1 Billion rows

In the diagram above, the arrows represent the flow of data through the system. As you can see, most of the query evaluation in the example is done by a single compute node. The system generated a data movement of 100 million rows, and only a single node performed additional filtering and aggregate count.

It's an improvement over a shared disk system, but it still has some serious limitations. Such a system could be well suited for simple key access (i.e. query touches a few specific rows), but any more complexity will generally result in poor performance

As with the shared disk system, adding more nodes will not help improve single query execution, and queries which operate over large volumes of data have the potential to saturate the message bus between the nodes.

Distributed Database Architectures: Shared Disk

2011-11-04T14:51:00.000-07:00

One of the most common questions I get about Clustrix is "How is Clustrix different than Database X?" Depending on who asks the question, Database X tends to be anything from Oracle RAC to MySQL Cluster. So I decided to put together a primer on different types of Distributed Database Architectures, and what each one means for real world applications.

Shared Disk Databases

Shared disk databases fall into a general category where multiple database instances share some physical storage resource. With a shared disk architecture, multiple nodes coordinate access to a shared storage system at a block level.

Adding clustering to a stand alone database through a shared disk cache.

Several of the older generation databases fall into this category, including Oracle RAC, IBM DB2 pureScale, Sybase, and others.

These systems all started out as single instance databases. The easiest way to add clustering to an existing database stack would be to share the storage system between multiple independent nodes. All of these databases already did paged disk access through a buffer manager cache. Make the caches talk to each to manage concurrence, and bam, you have a distributed database!

Most of the database stack remains the same. You have the same planner, the same optimizer, the same query execution engine. It's a low risk way to extend the database to multiple nodes. You can add more processing power to the database, keep data access transparency in the application layer, and get some amount of fault tolerance.

But such systems also have serious limitations which prevent them from getting very wide spread adoption, especially for applications which require scale. They simply don't scale for most workloads (footnote: see Scale w/ Sharing below), and they are extremely complex to administer.

Almost every new distributed database system built in the last 10 years has embraced a different architecture, mainly because the shared disk model has the following problems:

Cache coherence overhead
Extensive data movement across the cluster
Centralized query execution (only a single node participates in query resolution)

Cache Coherence and Page Contention

Let's assume the following workload as the first example:

DB 1: UPDATE mytable SET mycol = mycol + 1 WHERE mykey = X

DB 2: UPDATE mytable SET mycol = mycol + 1 WHERE mykey = Y

Now let's further assume that some of the keys for that update statement will share a page on disk. So we don't have contention on the same key, but we do have some contention on the same disk page.

Page contention in shared-disk systems.

Both nodes receive similar update queries. Both nodes must update the same physical page. But in order to do it safely, they must insure that all copies of the page are consistent across the cluster.

Managing such consistency comes at a cost. Consider some of the requirements which have to be satisfied:

Every node must acquire a page lock in order to read or write from the page. In a clustered environment, such locking results in communication between nodes.
If a node acquires a write lock on a page and modifies it, it must notify any other node to invalidate their caches.
If a node gets a page invalidation request, it must re-fetch the latest copy of the page, which also results in more network communication.
Each node caches the contents of the entire data set. It means that the effective cache size of the cluster is as large as the cache on any of the nodes. Adding more nodes does not scale cache efficiency.

Now imagine adding more nodes to your cluster. You end up with a system which has non-linear scaling in message complexity and data movement. It's common to see such systems struggle beyond 4 nodes on typical OLTP workloads.

Data Movement

Consider another example which poses problems for shared disk systems:

SELECT sum(mycol) FROM mytable WHERE something = x

Let's assume that mytable contains 1 Billion rows and that the something = x leaves us 1,000 rows. With a shared disk system, if the predicate results in a table scan, the entire contents of the 1B row table must be transferred to the node evaluating the query!

And from the previous example, we see that it's not just a matter of data movement across the SAN infrastructure. The system must also maintain cache coherence, which means lots of cache management traffic between all the nodes. Such queries on large data sets can bring the whole cluster to its knees.

Centralized Query Resolution

It's also worth while to point out that only a single node within the cluster can participate in query evaluation. So even if you throw more nodes at your system, it's not generally going to help you speed up that slow query. In fact, adding more nodes may slow down the system as the cost of managing cache coherence increases in a larger cluster.

Scaling with Sharding

It's possible to scale with shared disk, but it requires an extensive engineering effort. The application is written (a) to keep affinity of data accesses to nodes and (b) to keep each shard within its own data silo. You lose transparency of data access at the application level. The app can no longer send any query to any one of the nodes, or it will result in a catastrophic performance problem.

While you end up with a very expensive means of implementing sharding, there is some advantage to this approach over regular sharding. In case of node failure, you can more easily bring in a host standby without having to keep a dedicated slave for each shard. But in practice, getting fault tolerance right with a SAN based shared disk infrastructure is no easy task -- just ask anyone who manages SANs for a living.

Percona Evaluates Clustrix and MySQL

2011-10-25T08:34:00.001-07:00

Percona ran a Percona-written TPCC benchmark against Percona's MySQL Server on an Intel- SSD based machine, Percona MySQL Server on a FusionIO based machine, and a variety of different sizes of Clustrix's database clusters running Clustrix's Sierra relational database.

The results show that Clustrix:

Scales linearly from 3 to 9 nodes.
431% faster than the tested FusionIO system.
Tested system configured with 2x redundancy for Fault Tolerance.

Find the complete results and testing methodology in the Percona-Clustrix TPCC Evaluation.

There are a few conclusions that can be drawn from these results.

Scalability

First is that the Clustrix database really is scalable. If you look at the median throughput graph on page 7, the sixnode results are two times the three- node results and the nine- node results are three times the three-node performance at concurrency. This is exactly the behavior you need to see as a website or similar application scales. As a website gets more and more users, concurrency goes up and aggregate performance requirements continue to grow. This is in stark contrast to the MySQL-based systems tested, which followed the traditional performance curve up to their peak, and then down toward unresponsiveness as the concurrency increased.

Performance

The second conclusion we draw from this report is that performance exhibited by Clustrix is also more consistent than the MySQL systems. If you look at Appendix A, you can see graphs representing the instantaneous performance throughout the test. Clustrix maintains more consistent performance and lower response times even at high concurrency -- exactly when your website needs it most.

Redundancy and Fault Tolerance

A critical thing to note about this comparison is that the MySQL based systems have no redundancy. They either have a single FusionIO card or SSD drives in a RAID 0 configuration. If any component in the system fails (drive, motherboard, memory, etc.), the entire database will fail. In the Clustrix systems, fault tolerance is built in -- all data in stored redundantly and there is no single point of failure. The only way with MySQL to get a similar level of fault tolerance (but still not as seamless as Clustrix) is to have two servers with replication. Generating a replication feed would further slow down the MySQL systems, so we left them unencumbered by replication to get the best possible numbers.

Complex Workload that Scales

The most interesting conclusion, perhaps, is the more general statement that transactional, relational, full-featured SQL can scale and Clustrix has proved it in this testing and at multiple production customer deployments around the world. The TPCC test is a relatively complicated workload. There are multi-statement transactions, joins, aggregates, foreign key constraints, and a large percentage of writes.

These are exactly the things the NoSQL movement has proclaimed as too difficult to scale. They couldn't make true transactions scale so they introduced eventual consistency. They couldn't make relational calculus scale so they forced denormalization and complicated logic onto the app developers. NoSQL "solutions" have thrown all this out and the one bone they throw the app developers is the concept of a "flexible data model."

That's not a revolution, that is a feature and a simple one at that (See our co-founder’s post on how this can easily be done in a SQL database and how Clustrix has implemented it). You don't have to give up transactions or ACID properties, relational semantics, or any of the rich SQL features in the name of performance or scale. SQL will perform and scale just fine.

Use Cases: TheLadders and iOffer

2011-07-15T15:15:00.000-07:00

Earlier this year we set out to create use cases with several of our customers. The two companies that we have chosen to release today are TheLadders and iOffer.

TheLadders.com is the world’s largest online job search service that lists jobs with annual salaries of $100,000 or more. TheLadders were in search of a database solution that was highly scalable, fault-tolerant and allowed the team to avoid sharding.

iOffer is the fastest growing destination for interactive social commerce with a vibrant global community connecting visitors from over 190 countries in every language via millions of item listings. iOffer was trying to find a database that was MySQL- compatible, and did not require changes to their existing architecture, database schemas or applications.

Take a look and see how Clustrix has solved these companies database problems.

Percona Live NYC: Keynote

2011-06-20T11:12:00.000-07:00

Check out Sergei Tsarev's keynote, "Why SQL Wins," from the Percona Live NYC Conference.

Sergei takes a historical look at database architectures and their current trends. He explores why NoSQL is not the answer to our current problems, and outlines why we should not throw away decades of relational database research and development.

New Website

2011-06-16T14:46:00.001-07:00

We refreshed our website design! We're still working on getting more content for it, so look for changes in the upcoming weeks!

Upcoming Events

2011-03-25T11:04:00.000-07:00

Please join Clustrix, Inc. at the Web 2.0 Expo next week on March 28-31, 2011 in San Francisco, CA. Stop by our booth, #630, and test out our fault tolerant failure demonstrations in the Exhibit Hall. In addition to our live demo, we will be giving away cool prizes throughout the day and a big prize at the end of each day.

CEO Paul Mikesell will be handing out the big giveaway at approximately 2:30pm at the end of each expo day, must be present to win.

Clustrix will also be attending the MySQL Conference & Expo on April 11-14, 2011 in Santa Clara, CA as a Gold Sponsor.

Be sure to mark your calendars and stop by our booths because there are “no limits” to your opportunities with Clustrix.

Profile Driven Performance Optimization

2011-02-28T09:49:00.000-08:00

Recently, one of our support engineers noticed that a customer cluster got a little sluggish during a routine maintenance operation. In looking back at historical data, we noticed that the cluster saw a decrease in transaction throughput. The system should have done its cleanup in the background at a low priority, but that didn't happen. I thought it would be interesting to describe our process for addressing a performance problem like this one. I will also share the tools we developed and used.

Reproduce, Measure, and Automate

The symptom our support engineer noticed was a "sluggish" system. More accurately, we saw that query latencies increased and overall system throughput dropped. So our first step was to reproduce the problem under some approximation of the customer's workload. It was a bit tricky to get all of the prerequisite conditions just right, but we finally landed on the right set of circumstances.

Once we reproduced the issue, the next step was to automate the test to ensure that we could consistently demonstrate the problem. It boiled down to using one of our standard performance harnesses with a couple of tweaks.

The Profile

Our most commonly used profiling tool is based on the Linux oprofile infrastructure. But we had to modify it to better suit our system. Why? Because substantial portions of our codebase are written in an event driven/continuation passing style programming model. Call graph reporting in ofprofile is highly dependent on the stack, and the C stack often does not contain enough information to provide a valuable analysis in our system.

To get around this problem, we modified oprofile to gather additional information from the system. The key concept was adding a method for tagging a series of continuation calls and getting oprofile to sample these tags. The tag itself includes enough information to map the executing code to a logical system module hierarchy. The screen shot below shows an example output from one of our reporting tools.

Coming back to our original problem, the screen shot shows the output of the profile analysis tool during the dip in performance. The tool allows us to isolate the report to a specific processor. In this case we're looking at the output for our management core.

Interpreting the results we see that:

The execution engine dominates performance, and most of that is in the storage engine
The storage engine spends 63% of all its cycles on CRC32 calculations
Our particular maintenance task was erroneously bound to the management core
The task should have a lower priority

CRC Performance

The root cause of the performance degradation was a scheduling issue. However, it caused me to take a closer look at our crc algorithm. We use the crc implementation from zlib, and it should be reasonably well optimized. Whipping up a quick benchmark, I we see that the zlib implementation takes about 41us to checksum a 32k block. In the test, we were reading about 224MB/s per node. That's 293ms of checksums in a second. No wonder we saw a performance drop.

I looked around and found a paper from Intel describing their slice-by-eight crc approach. A quick benchmark revealed that the same 32k checksum could be done in 27us -- a 37% percent improvement over zlib's version.

With Nehalem-based processors, Intel introduced an SSE instruction for a hardware based crc computation. I haven't had a chance to conduct my own benchmarks, but the following results suggest that we could get a 2-3x speedup over slice by eight.

Source Code

I'm releasing the source for the tools referenced in this post. The user space oprofile daemon replacement requires our database runtime which I am not making public.

Clustrix as a Document Store: Blending SQL and JSON Documents

2011-02-07T07:50:00.000-08:00

Many responses to my previous post claimed that my comparisons to MongoDB were unfair because MongoDB is a "document store" and Clustrix is a SQL RDBMS. Somehow that distinction made almost any comparison invalid. In response, I spent my weekend coding up a prototype document store interface for Clustrix to demonstrate that a different data model is not in itself an architectural differentiator.

At first I thought of just creating a different front end for Clustrix; our architectural model makes this easy. However, I quickly decided against it because in many ways, it would be much more limiting than a SQL based interface (e.g. joins, flexibile aggregation, subqueries, etc.)

Instead, I extended our SQL syntax to support native operations on JSON objects. So now you can do the following:

clustrix> create table files (id int primary key auto_increment, doc json);
Query OK, 0 rows affected (0.04 sec)

clustrix> insert into files (doc) values ('{"foo": {"bar": 1}, "baz": [1,2,3,4]}');
Query OK, 1 row affected (0.00 sec)

clustrix> select id, files.doc::foo.baz from files where files.doc::foo.bar = 1;
+----+--------------------+
| id | files.doc::foo.baz |
+----+--------------------+
| 1 | [1,2,3,4] |
+----+--------------------+

1 row in set (0.00 sec)

clustrix> create index foo_bar on files (doc::foo.bar);

Query OK, 0 rows affected (0.08 sec)

The database has native support for dealing with JSON documents. We're not simply storing text blobs inside of some column and getting them back. We're exposing the contents of the JSON document to the underlying planner and execution engine. Immediately, we get the following advantages:

Ability to do joins across document collections
Extremely powerful and flexible query language
Ability to index into JSON objects
Transactional semantics built-in

Taking it alittle further, I added a select clause modifier which instructs the database to return all row data as a JSON document, including fields which come from "a relational column." The following example shows how the database can seamlessly join between our json data type and other data types in the system, and then return the result as a JSON objects.

clustrix> select _json_ f.doc, u.doc::username
    ->   from files f
    ->   join users u on f.doc::user_id = u.user_id
    ->  where f.doc::foo.bar = 1\G
*************************** 1. row ***************************
json: {"f.doc": {"foo": {"bar": 1}, "baz": [1,2,3,4]}, "u.doc::username": "sergei"}
1 row in set (0.01 sec)

We can continue to extend the syntax. For example, adding operators to manipulate lists within JSON. Or adding optional schema checking for contents of the JSON (i.e. something along the lines of DTD for XML). I'm sure you can think of more.

An any case, one can build a system which combines the best characteristics of a document store with the power of SQL. Both models can coexist in the same database, allowing the devloper to trully choose the data model which best suits his or her needs.

Sierra and the Clustrix Database Stack

2011-02-02T15:53:00.000-08:00

A lot of the responses to my previous posts criticized my choice of comparing a document database like MongoDB to a relational database like Clustrix. I tried to examine aspects of the database architecture which have nothing to do with the data model, but somehow the data model would always come up. There is a set of concerns that's common to all database systems, whether you are a document store or a relational database.

But first, think about how your database would handle the following workload. Chose whatever data model you find best suited for my use case.

get me all the records from foo where a = ? and b = ?

foo has an index over A and another index over B.
A and B have non-uniform data distributions. A is 90% value "X" and B is 90% value "Y".
We have 1 billion records in foo.
The database has a choice: index A, index B, or scan and filter.
50% of the queries are a = X and b = Z, the other 50% are a = Z and b = Y

What does your database do?

If the database always chooses index A or index B for all queries, then it ends up examining 900,000 rows 50% of the time.

We'll come back to my question later. First, I will briefly describe what the database stack in Clustrix looks like. I need to set a foundation so that my next set of posts makes sense.

To anyone who has experience with DBMS systems, the stack should look very familiar. We have fairly strict abstraction of interfaces between the various portions of the stack.

The Protocol Handler and Query Parser are responsible for taking in user connections and translating SQL into our intermediate representation called Sierra. We can actually support multiple type of dialects at these two layers; the constraint is that we must be able to express the query language in Sierra. And Sierra is much more expressive than SQL. The value of Sierra is that it provides an extensive planner framework for reasoning about distributed database queries.

So the Planner/Optimizer only accepts Sierra. It runs through a search space of possible plans and prunes them based on cost estimates. After coming up with the best plan candidate, it translates into another intermediate representation used by the Distributed Compiler, which reasons out the physical execution plan for our query. Finally, we compile the query into machine code and execute it.

As a performance optimization, we cache the compiled programs and plans. Clustrix does not need to optimize and compile every query, and you don't need to use prepared statements to get this behavior.

Back to my question. What did you come up with? Well, I can tell you what happens in Clustrix. During the planning phase of Sierra, we examine the various statistics that the database keeps about the data distribution in our indexes. The statistics include tracking:

Number of distinct values across index columns
Quantile distributions
Hotlist tracking top n values within a column

Clustrix correctly chooses the plan which will result in the least number of rows examined for all input parameters.

Now, don't get me wrong. I am not claiming that data distribution statistics are unique to Clustrix. On the contrary, they are very common in any modern RDBMS. I'm using it as an example of a requirement that's independent of any data model, and it's actually a very important feature to have.

MongoDB vs. Clustrix Comparison: Part 2

2011-02-01T10:01:00.000-08:00

Introduction

In Part 1 of my comparison, I ran some performance benchmarks to establish that relational systems can scale performance. In this post I would like to focus more on the High Availability and Fault Tolerance aspects of the two systems. The post will go over the approach of each system and what it means for fault tolerance and availability. I will also conduct a test of the claims: I'm going to fail a node by pulling power to see what happens.

A Primer on Clustrix Data Distribution

Clustrix has a fine grained approach to data distribution. The following graphic demonstrates the basic concepts and terminology used by our system. Notice that unlike MongoDB (and many other systems for that matter), Clustrix applies a per-index distribution strategy.

There are many interesting implications for query evaluation and execution in our model, and the topic deserves its own set of posts. For the curious, you can get a brief introduction to our evaluation model from our white paper on the subject. For this post, I'm going to stick to how our model applies to fault tolerance and availability.

You can find documentation for MongoDB's distribution approach on their website. In brief, MongoDB chooses a single distribution key for the collection. Indexes are co-located with the primary key shard.

Clustrix Fault Tolerance and Availability Demo

Fault Tolerance

Both Clustrix and MongoDB rely on replicas for fault tolerance. A loss of a node results in a loss of some copy of the data which we can find elsewhere in the system. The MongoDB team put together a good set of documentation describing their replication model. Perhaps one of the most salient differences between the two approaches is the granularity of data distribution. The unit of recovery on Clustrix is the replica (a small portion of an index), while the unit of recovery in MongoDB is a full instance of a Replication Set.

For Clustrix, this means that the reprotection operation happens in a many-to-many fashion. Several nodes copy small portions of data from each of their disks to several other nodes onto many disks. The advantages of this approach are:

No single disk in the system becomes overloaded with writes or reads
No single node hotspot for driving the reprotect work
Incremental progress toward full protection
Independent replica factors for each index (e.g. primary key 3x, indexes 2x)
Automatic reprotection which doesn't require operator intervention
All replicas are always consistent

One of the interesting aspects of the system is the complete automation of every recovery task. It's built in. I don't have to do anything to make that happen. So if I have a 10 node system, and a node fails, in an hour or so I will have a completely protected 9 node system without any operator intervention at all. When the 10th node comes back, the system will simply perceive a distribution imbalance and start moving data back onto that node.

While the Replica Sets feature in MongoDB is nicer than replication in say MySQL, it's still highly manual. So in contrast with the above list for Clustrix, for MongoDB we have:

Manual intervention to recover from failure
The data is moved in a one-to-one fashion
All data within a Replica Set has the same protection factor
Failures can lead to inconsistency

Availability

Both systems rely on having multiple copies of data for Availability. I've seen a lot of interesting discussion recently about the CAP theorem and what it means for real-world distributed database systems. It's another deep topic which really deserves its own set of posts, so I'll simply link to a couple posts on the subject which I find interesting and illuminating:

At Clustrix, we think that Consistency, Availability, and Performance are much more important than Partition tolerance. Within a cluster, Clustrix keeps availability in the face of node loss while keeping strong consistency guarantees. But we do require that more than half of the nodes in the cluster group membership are online before accepting any user requests. So a cluster provides fully ACID compliant transactional semantics while keeping a high level of performance, but you need majority of the nodes online.

However, Clustrix also offers a lower level of consistency in the way of asynchronous replication between clusters. So if you want to setup a disaster recovery target in another physical location over high-latency link, we're able to accommodate that mode. It simply means that your backup cluster may be out of date by some number of transactions.

MongoDB has relaxed consistency all around. The Replication Set itself uses an asynchronous replication model. The MongoDB guys are upfront about the kinds of anomalies they expose. The end user gets the equivalent of read uncommitted isolation. Mongo's claim is that they do this because they (1) can achieve higher performance, and (2) "merging back old operations later, after another node has accepted writes, is a hard problem." Yes. Distributed protocols are a hard problem, but it doesn't mean you should punt on them.

Availability Continued

There's also a more nuanced discussion to availability. One of the principal design features of Clustrix has been to aim for lock-free operation whenever possible. We have Multi-Version Concurrency Control (MVCC) deeply ingrained in the system. It allows a transaction to see a consistent snapshot of the database without interfering with writes. So a read in our system will not block a write.

Building on top of MVCC, Clustrix has implemented a transactionally safe, lockless, and fully consistent method for moving data in the cluster without blocking any writes to that data. All of this happens completely automatically. No administrator intervention required. So when the Rebalancer decides to move a replica from Node 1 to Node 3, the replica can continue to take writes. We have a mechanism to sync changes to the source replica with the target replica without limiting the replica availability.

Compare that to what many other systems do: read lock the source to get a consistent view for a replica copy. You end up locking out writers for the duration of the data copy. So while your data is available for reads, it is not available for writes.

After a node failure (or to be more precise, a replica failure within a set), MongoDB advocates the following approach:

Quiesce the master (read lock)
Flush dirty buffers to disk (fsync)
Take an LVM snapshot of the resulting files
Unlock the master
Move the data files over to the slave
Let the slave catch up from the snapshot

So a couple of points (a) the MongoDB is not available for writes during steps (1) and (2), and (b) it's a highly manual process. It reminds me very much of the MySQL best practices for setting up a slave.

Conclusion

I've seen a lot of heated debates about Consistency, Availability, Performance, and Fault Tolerance. These issues are deeply interconnected and it's difficult to write about any of them in isolation. Clustrix maintains a high level of performance without sacrificing consistency and very high degree of availability. I know that it's possible to build such a system because we actually built it. And you shouldn't sacrifice these features in your application because you believe it's the only way to achieve good performance.

MongoDB vs. Clustrix Comparison: Part 1 -- Performance

2011-01-30T19:55:00.000-08:00

UPDATE:

You can now download the benchmark code. As mentioned in my post, I populated both databases with 300M rows. I played around with the best client/host/thread ratio for each database to achieve peak throughput. I used MongoDB v1.6.5.

For the MongoDB read tests, I used the read-test.cpp harness. I didn't have time to do a proper getopt parsing for it, so I would modify it by hand for test runs. But it's very straight forward.

The C version of the MySQL/Clustrix harness is not included because I used a Clustrix internal test harness framework -- to save time. It doesn't impart Clustrix any advantage in the test, and it relies too much on our infrastructure to be of any value. You can still use the Python based test harness -- it just requires a lot of client cpu power.

UPDATE 2:

There's an interesting conversation over at Hacker News about this post.

Introduction

With all the recent buzz about NoSQL and non-relational databases, the marketing folks at Clustrix asked a question: Do we have the right solution for today's market? It's a fair question, especially since Clustrix is a fully featured RDBMS with a SQL interface. And we all heard that SQL doesn't scale, right?

So that brings us around to the next question: What do people actually want out of their database? Surely it's not simply the absence of a SQL based interface. Because if that's the case, Berkeley DB would be a lot more popular than say SQLite. Over the years, we've had many conversations with people about their database needs. Over and over, the following has always come up a list of must have features for any modern system:

Incrementally Scalable Performance
High Availability and Fault Tolerance
Ease of Overall System Management

Interestingly enough, we never heard that SQL or the relational model was the root of all their problems. It appears that the anti-SQL sentiment came around with this sort of false reasoning.

I have a SQL based RDBMS.
I can't seem to scale my database beyond a single system.
Therefore SQL is the problem.

The NoSQL movement embraced this reasoning. The NoSQL proponents began to promote all sorts of "scalable" systems at the expense of venerable DBMS features like durability. And they kept going. What else don't we need? Well, we don't need Consistency! Why? Because that's really hard to do and keep performance. Slowly but surely, these systems would claim to have a panacea for all of your scalable database needs at the expense of cutting features we've come to expect from 40 years of database systems design.

Well, that's just bullshit. There is absolutely nothing about SQL or the relational model preventing it from scaling out.

Over the next set of posts, I'm going to compare MongoDB and Clustrix using the above evaluation criteria: Scalable Performance, Availability and Fault Tolerance, and Ease of Use. I am going to start with Performance because no one believes that you can grow a relational database to Internet Scale. And to put the results into context, I chose to compare Clustrix to MongoDB because (1) it doesn't support SQL, (2) it can transparently scale to multiple nodes, and (3) it seems to be the new poster child for NoSQL.

Performance

Conducting performance benchmarks is always challenging. First, you have to decide on a model workload. Next, you have to accurately simulate that workload. The system under test should be as close as possible to your production environment. The list gets long. In the end, no benchmark is going to be perfect. Best you can hope for is reasonable.

So I looked at some common themes in the workloads from some of our customers, and decided that I would simulate a basic use case of keeping metadata about a collection of 1 Billion files. Whether you're a cloud based file storage provider or a photo sharing site, the use case is familiar. The test would use the appropriate access patterns for the database. Since MongoDB does not support joins, I'm not going to put it at a disadvantage by moving join logic into the application. Instead, I'm going to make full use of the native document centric interface.

Benchmark Overview

10 node cluster of dedicated hosts (SSD, 8 cores/node, 32GB ram/node)
2x replication factor
A data set containing information about ~~1 Billion files~~ 300 Million (see bellow)
A read only performance test
A read/write performance test
Both databases will use the exact same hardware

The test uses the following schema:

CREATE TABLE files (

    id                            bigint     NOT NULL,

    path                         varchar    NOT NULL,

    size                         bigint     NOT NULL,

    server_id               int        NOT NULL,

    deleted                   smallint   NOT NULL,

    last_updated         datetime   NOT NULL,

    created                   datetime   NOT NULL,

    user_id                   bigint     NOT NULL,

    PRIMARY KEY (id),

    KEY server_id_deleted (server_id, deleted),

    KEY user_id_updated (user_id, last_updated),

    KEY user_id_path (user_id, path)

);

Additionally, the data set has the following characteristics:

path is a randomly generated string between the length of 32 and 128 characters
server_id has a distribution of 0-32
deleted has 1% value 1, and the rest 0 (lumpy data distributions tests)
for MongoDB, we use user_id as the shard key

Test 1: Loading the Initial Data Set

The test harness itself is a multi-host, multi-process, and multi-threaded python application. Because of the GIL in python, I ended up designing the test harness so that it forks off multiple processes, with each process having some number of threads. It also turns out that I needed more than 10 client machines to saturate the cluster with reads using python, so I rewrote the read tests suing C++ for MongoDB and C for Clustrix.

While populating the dataset into MongoDB, I kept on running into a huge drop off in performance at around 55-60M rows. A 10 node cluster has an aggregate of 320GB or ram and 80 160GB SSD drives. That's more than enough iron to handle that much data. As I started to dig in more, I saw that 85% of the data was distributed to a single node. MongoDB had split the data into multiple chunks, but its balancer could not (would not?) move the data to other nodes. Once the database size exceeded that node's available memory, everything went to shit. The box started thrashing pretty badly. It seems that under a constant high write load, MongoDB is unable to automatically redistribute data within the cluster.

To get the test going, I split the files collection into an even distribution. Without any load on the cluster, I watched MongoDB move the chunks onto the 10 replica sets for an even layout. Now I was finally getting somewhere.

Immediately, I noticed that MongoDB had a highly variable write throughput. I was also surprised at how low the numbers were. Which led me to discover Mongo's concurrency control: a single mutex over the database instance. Furthermore, in tracking the insert performance along with memory utilization, I could see that getting to more than 300 million records would spill some of the data set to disk. While that's a reasonable benchmark for a database, I decided that I would keep the data set memory resident for Mongo's sake.

The drop-off on the Clustrix happened because not all of the load scripts finished at the same time. A couple of the client nodes were slower, so they took a bit longer to finish up their portion of the load.

For a system which eschews consistency and durability, the write performance on MongoDB looks atrocious. Initially, I thought that Mongo completely trashing Clustrix on the write performance. The result was a complete surprise. Here's why I think Clustrix did so much better:

Highly concurrent b-tree implementation with fine grained locking
Buffer manager and transaction log tuned for SSD
Completely lock-free Split and Move operations (very cool stuff, another post)

Conversely, Mongo did so poorly because it:

Has a big fat lock, severely limiting concurrency
It relies entirely on the kernel buffer manager

Total time to load was 0:37 (hh:mm) on Clustrix and 4:47 on MongoDB.

Clustrix was 775% faster for writes than MongoDB!
And that's with fully durable and fully consistent writes on the Clustrix side.

Test 2: Read Only

The read test consists of the following basic workloads:

get the 10 latest updated files for a specific user
count the number of deleted files on a given server id

I chose these queries because they are representative of the types of queries our example application would generate, and they are not simple point selects. Getting a distributed hash table working is easy. But DHTs tend to fall apart fairly quickly when queries start introducing ordering, examining multiple rows, or other non key-value lookups. In other words, real-world use.

MongoDB
C++ test harness. Peak throughput at 64 concurrent client threads.

db.files.find({user_id: user_id}).sort({last_updated: -1}).limit(10)

55,103 queries/sec

db.files.find({'server_id': server_id, 'deleted': 1}).count()

675 queries/sec

Clustrix
C test harness. Peak throughput at 256 concurrent client threads.

select * from benchmark.files where user_id = .. order by last_updated desc limit 10

56,641 queries/sec

select count(1) from benchmark.files where server_id = .. and deleted = 1

625 queries/sec

So on a read-only test, MongoDB and Clustrix are within 1% of each other for test1. Clustrix is faster on test 1 and MongoDB is 7% faster on test2. I captured a profile of Clustrix during a test1 run, and saw that the the execution engine dominates CPU time (as opposed to say SQL parsing or query planning). In looking at the profiles during test2 runs on Clustrix, I saw that we had a bunch of idle time in the system, so there's room for optimization.

But real-world loads tend to be read/write, so let's see how Mongo does when we add writes to the equation.

Test 3: Read/Write

My initial plan called for a combination of read-centric and write-centric loads. It seems that most web infrastructures are heavier on the reads than writes, but there are many exceptions. In Clustrix, we use Multi-Version Concurrency Control, which means that readers are never blocked by writers. We handle both read heavy and write heavy workloads equally well. Since MongoDB seems to do much better with reads than writes, I decided to stick to a read-centric workload.

The Clustrix test shows show very little drop off in performance for reads. On the Mongo side, I expected to see a drop off in performance directly proportional to the amount of write load.

However, what I saw mind blowing: Mongo completely starved the readers! The following graph shows the query load on one of the 10 shards during the write portion of the test. I simply started up a single write thread while letting the read test run. The write was active for all of 60 seconds, and it took Mongo an additional 15 seconds to recover after the writer stopped.

MongoDB
FAIL

Custrix
test1: 4,425 writes/sec and 49,099 reads/sec (total 53,524 queries/sec) (92% read / 8% write)
test2: 4,450 writes/sec and 625 reads/sec

The test2 aggregate query is much more computationally expensive compared to an insert. So the read/write ratio for test2 became very skewed. Note that Clustrix did not drop in read throughput at all.

Overall, you can see why every modern DBMS choose to go with the MVCC model for concurrency control.

Conclusion

The SQL relational model can clearly scale. The only place where MongoDB could compete with Clustrix was on pure read-only workloads. But that's just not representative of real world application loads

Building a scalable distributed system is more about good architecture and solid engineering. Now that we have scale and performance out of the way, I'm going to review the other important aspects of a DBMS in my upcoming posts.

SQL Is Not The Problem

2010-10-22T11:02:00.002-07:00

There is a lot of buzz about NoSQL out there. According to the hype, NoSQL solves scalability issues, NoSQL simplifies development, NoSQL has unlimited write performance, etc. While these claims may technically be true, they are in no way exclusive to the NoSQL offerings out there. All of these things are achievable in a SQL database with full ACID and relational capabilities. In a recent post, Derrick Harris asks “Will Scalable Data Stores Make NoSQL a Non-Starter?”. He Makes the point that scalable SQL is a reality today so why would someone go to a NoSQL solution? What benefit could it possibly bring? Would anyone choose to give up transactions if they didn’t have to? Would anyone choose to give up ACID properties? Would anyone choose to give up the ability to do relational operations? How would giving up any of these things simplify development? Looking at the comments on Derrick’s post provides a clue. They mention the “flexible schema” as the key feature.

A flexible schema is defined by the ability to assign arbitrary properties to an object without having to have defined columns for those properties. For example, in a single table, object A could represent a picture and have “height” and “width” properties while object B could represent an audio stream that has “length” and “bitrate” properties. In a traditional RDBMS environment, you’d probably create multiple tables for each different object type which can be inconvenient and possibly inefficient.
Is that it? Why wouldn’t you just add that feature to a scalable SQL database rather than throw away all the good things SQL databases have to offer? You could easily add a “map” column to a table that allows you to store an arbitrary map of key/values associated with an object. You now have the tools to implement a flexible schema while maintaining relational capabilities, ACID properties, and full scalability. It’s the best of both worlds.

Surprising Advice For Startups

2010-06-25T10:58:00.001-07:00

Clustrix just exhibited at two good shows: Structure and Velocity. I spent my time at Structure. One of the highlights of the show for Clustrix was the The Future of SQL In the Cloud panel that Paul was on. There was some good discussion about SQL vs. NoSQL and different products available for scaling your database. The title is a bit of a misnomer, however. Database challenges are orthogonal to the cloud. The database is a critical challenge no matter how the application is deployed.

But this isn’t what surprised me the most at Structure. That came in the How Does a Company Scale in Real Time? panel. This panel had representatives from PayPal, Engine Yard, Yahoo!, Facebook, and Zynga. These panelists are all responsible for ensuring their products continue to function smoothly. These companies are all recognizable and successful. I was really hoping to hear some good insight into operating large scale web properties and perhaps some good advice for new web startups. I think the money question came at around 16:30 in the video. Jonathan asks the panelists to rewind the clock and say what they would do differently and to give advice to startups for what they should focus on to avoid application bottlenecks. There was a lot of advice offered. There was talk of establishing a single-signon infrastructure. There was emphasis on avoiding SQL entirely, saying it is far and away the biggest scalability problem in web architectures today. There was a suggestion you should invest up front in separating out your architecture. You should think hard about the abstraction layers in key parts of your system and get the infrastructure right first. Another talks about getting the right instrumentation and metrics in your application. You should put in the right levels of abstraction, and the right levels of caching. You need to force yourself to run your application on two servers. Think about sharding and replicating at the beginning and force your scaling challenges to come up in advance before you have to scale.

None of this advice resonated with me. Startups have enough challenges finding the right business model and securing the initial set of customers. Why would you spend your very limited money, and more importantly, time on infrastructure and architecture of the back-end before you’ve proved your business? Matthew Mengerink, VP of Customer Quality, Engineering Services, and Site Operations at PayPal finally stepped in and made some sense. He says he would do nothing different. To him, a startup working on architecture is a waste of a dollar. He advises spending your time and money on making the business model work. Thank you, Matthew!

This discussion reminded me of a post titled Start In the Middle I read a while ago. In there, it offers a bit of advice that should be obvious that is so often not: solve the interesting bit of the problem first. Prove there’s value in the core idea, and then flesh out the infrastructure around it. Do what makes your business unique first and put the majority of time into that. Delay building the rest of the surrounding architecture until its really needed or when possible, just buy those bits. The rest of that BS is not what your customers see and not what makes you money.

So what’s my advice? Make the code and architecture as simple as you possibly can but no simpler. Don’t abstract too early and don’t make premature optimizations. Rewriting code and adding abstractions later is not a sign of coding failure, rather it’s a sign the product is successful. Choose the data model that fits your problem. That is frequently SQL. Full transactional and relational SQL is exceptionally expressive and it fits many problems so well. Why wouldn’t you use it? Don’t re-invent things that are not part of your core business. When the product becomes successful and the load starts to ramp, collect hard data to find the bottlenecks and fix them when they become a problem. When your data shows that your database is that bottleneck, perhaps Clustrix can help.

Not All Clusters Are the Same

2010-04-29T10:56:00.001-07:00

There are a wide variety of techniques out there for clustering storage appliances. The question is: what problem are you really trying to solve? If you look at Isilon’s clustered storage appliances (where I was chief architect), you’ll see that the clustering is done at the block level. The block addresses are generalized into a generic (node, drive, block_num) tuple and the on-disk data structures simply use that generalized address everywhere a block address would normally be used (plus a bunch of details I’m glossing over). The communication on the back end of an Isilon cluster is block reads and writes, transaction messages, and lock messages (plus some other miscellaneous bits). Each read or write operation is controlled by the initiator, and the smallest granularity of locking is at the block level. Cache lives both at the disk and at the initiator. If you were to put it into an architecture category, you’d call it an Infiniband SAN (Storage Area Network). This is perfect for a file system. This architecture lends itself to zero-copy, extremely high-performance file access for streaming files, very low CPU utilization on the nodes holding the disks (which allows the addition of the accelerator nodes for high speed FibreChannel and 10GbE), infinite scalability, and extremely low latency for operations on cached data.

However, it doesn’t support high read/write concurrency on a single file. Imagine if you ran an OLTP database with a high write load using an architecture like that. With the locking done at the block level, you can never expect to get high concurrency for items smaller than a block. Every node that wants to write to a block would have to get an exclusive lock on that block, which invalidates other nodes’ caches. If you had an active table with massive read/write load sitting on top of a cluster like this, performance would tank, dominated by lock contention. Then why do some databases take this approach to scale? How can you possibly make a shared-backend cluster resembling a SAN and expect it to scale with a database workload like some have done? How can you make an expandable storage engine plug-in and expect the entire database to scale? What works extremely well for a file system does not work at all for a database. We need a new approach.

Clustrix has a new approach. Rather than shipping the data blocks on the back end, we ship the queries. That may sound like an innocuous statement, but really it has a far-reaching impact on the architecture. To learn more, read the white paper A New Approach: Sierra Distributed Database Engine that I wrote on this subject. It shows how Clustrix has taken a novel approach to solve the clustered database problem, resulting in a database system that can handle high concurrency at any scale.

The Tyranny of the OR vs. the Genius of the AND

2010-04-28T10:53:00.001-07:00

Jim Collins and Jerry Porras first coined this phrase in Built to Last. In their book, they describe how choosing between seemingly contradictory concepts—focusing on this or that—leads to missed opportunities. Is the product low cost or high quality? Do I focus on short-term opportunities or long-term strategy? Should the company be bold or conservative? Built to Last is focused on business and what makes great companies continue to succeed year after year. Collins and Porras discovered that the best companies find a way to embrace the positive aspects of both sides of a dichotomy, and instead of choosing, they find a way to have both.

These same dichotomies exist in technology. Some common dichotomies include: low power vs. high performance, high stability vs. new features, and ease of use vs. flexibility. These are false choices. You can have high performance and low power—just look at the huge gains made through die shrinks. You can have new features without sacrificing stability with the right development process and proper QA. You can have an easy to use product that also enables maximum flexibility. All it takes is thoughtful design. We can embrace “the genius of the AND” here, too.

The tyranny of the OR still is strong in the database world. Do you want relational and ACID capabilities or do you want to be able to scale? An entire movement has been formed to answer this exact either-or question: the NoSQL movement. Digg has blogged about it (http://about.digg.com/blog/looking-future-cassandra and http://about.digg.com/blog/saying-yes-nosql-going-steady-cassandra), talking about how they did a large rewrite of their code to move to Cassandra, working around limitations in the process. If you look at the Digg blogs, they really did have a point. Once you scale beyond what a single box can do, you had two options: partition the database or move away from traditional relational databases. In both cases, you lose relational capabilities. You can either have scale or you can have the functionality you need.

They did have a point. Clustrix now offers a third option. The Clustrix database offers full relational and transactional capabilities and it can scale. The Clustrix product speaks the MySQL protocol on the wire (but uses no MySQL code) and allows seamless online scaling. It supports arbitrary relational calculus at any scale. It supports full ACID semantics—not eventual consistency—yet writes continue to scale. Clustrix has learned to embrace the genius of the AND.

How Clustrix was born

2010-04-15T10:00:00.000-07:00

Back in 2005 I was at Isilon. We had developed the first (and still the best) truly distributed scale-out NAS solution. We were on the road to an IPO and our customers were excited about our products and what we had done with them. I was on the phone with Jake, who worked for one of our largest customers. Jake said this to me: “This is great for what you guys have done of our basic file storage, but can you do anything about our databases?”. BAM! Just like that I got super excited about this idea and started looking around to see who else had these types of scalability and fault tolerance issues with their databases. As you probably know, it turns out that it was an issue for just about everyone.

I got together with Sergei Tsarev (Clustrix co-founder) and we started working on the problem. What is it about these databases that isn’t scaling? Storage usage and query performance. Ok, so why isn’t it scaling? Because the query processor is only as big as the box can be, and scaling up with bigger and bigger boxes is a non-starter (forklift upgrades, leaving the commodity price curve, etc.). We saw some systems out there doing things with virtualized clustered storage engines running behind traditional query planners, but those always failed to scale because you wind up pulling all of this data, as well as dealing with locking and concurrency, over the network. We realized that the only way to bring true scalability is to fan out the work as you grow the cluster.

As Aaron (our CTO) says – ‘Bring the query to the data, not the data to the query’. In his whitepaper, A New Approach: Sierra Distributed Database Engine, Aaron provides a nice description of how that concept became the seed for the revolutionary technology that is the heart of our Clustered Database Systems.