Saturday at MySQL Connect

The first day of the first MySQL Connect conference is done.  It's been a busy day!  Many attendees are interested in the new MySQL Server 5.6 release, but of course MySQL Cluster is the main draw here.  After a session from Oracle on the new features in 7.2, and early access features in 7.3, I attended Santo Leto's MySQL Cluster 'hands on lab'.  Despite having started more clusters than most, it felt like a new and exciting experience installing and running my cluster alongside everyone else.  The lab machines had some networking issues, but with Santo's help we seamlessly failed-over to some downloads he'd prepared earlier - very professional!

Afterwards it was my turn to talk on the subject of MySQL Cluster performance.  The quality of the questions was impressive - there seems to be a very capable crowd in San Francisco this weekend.  I was flattered that some audience members were taking pictures of my slides, but there's no need, you can download them here, and probably from Oracle somewhere too.

Finally we had a 'Birds of a Feather' session where a number of users asked questions about replication, backups, failover + arbitration, system recovery etc...  Great to get the chance to explain some things, hear about user experiences, and get some feedback about user experience.

There's been a lot of good content so far, and interesting discussions, but the highlight for me has been meeting with colleagues old and new, from MySQL, Sun and Oracle times.  More of the same tomorrow, I'd better get some sleep!

One billion

As always, I am a little late, but I want to jump on the bandwagon and mention the recent MySQL Cluster milestone of passing 1 billion queries per minute. Apart from echoing the arbitrarily large ransom demand of Dr Evil, what does this mean?

Obviously 1 billion is only of interest to us humans as we generally happen to have 10 fingers, and seem to name multiples in steps of 10^3 for some reason. Each processor involved in this benchmark is clocked at several billion cycles per second, so a single billion is not so vast or fast.

Measuring over a minute also feels unnatural for a computer performance benchmark - we are used to lots of things happening every second! A minute is a long time in silicon.

What's more, these reads are served from tables stored entirely in memory - and everyone knows that main memory is infinitely fast and scalable and always getting cheaper, right?

If we convert to seconds we are left with only 17 million reads per second! Hardly worth getting out of bed for?

On the contrary, I think that achieving 17 million independent random reads per second, each read returning 100 bytes across a network, from a database that also supports arbitrary SQL, row locking, transactions, high availability and all sorts of other stuff, is pretty cool. I doubt that (m)any other similar databases can match this raw performance, though I look forward to being proved wrong.

(Also, don't forget to meet + beat 1.9 million random updates/s, synchronously replicated)

Raw performance is good, but not everyone just needs horsepower. The parallel, independent work on improving join performance (also known as SPJ/AQL) and query optimisation helps more applications harness this power, by improving the efficiency of joins.

I wrote a post about SPJ/AQL at the start of last year, when it was still in the early stages. Since then much has improved, to the extent that the performance improvement factors have become embarrassingly high on real user queries. A further post on the technical details of SPJ/AQL is long overdue... Perhaps the most interesting details are on the integration between the parallel, streaming linked operations and the essentially serialised MySQL Nested Loops join executor. A linked scan and lookup operation can be considered to be a form of parallel hash join, which the normal MySQL NLJ executor can invoke as part of executing a query. Who says Nested Loop joins can't scale?

Eventual consistency with transactions

In my last post I described the motivation for the new NDB$EPOCH conflict detection function in MySQL Cluster. This function detects when a row has been concurrently updated on two asynchronously replicating MySQL Cluster databases, and takes steps to keep the databases in alignment.

With NDB$EPOCH, conflicts are detected and handled on a row granularity, as opposed to column granularity, as this is the granularity of the epoch metadata used to detect conflicts. Dealing with conflicts on a row-by-row basis has implications for schema and application design. The NDB$EPOCH_TRANS function extends NDB$EPOCH, giving stronger consistency guarantees and reducing the impact on applications and schemas.

Concurrency control in a single synchronous system

MySQL Cluster is a relational system. Data is stored in tables with defined schemas of typed columns. As with any relational system, real-world concepts can be modelled in a number of ways with different trade offs. One such consideration is the level of normalisation applied to a data model. Transactions and concurrency control ensure that some data modelled using multiple tables, rows and columns, appears to any external observer to move instantaneously between stable, self consistent states. This is a powerful simplification, and eases the complexity burden on application writers. Each transaction provides the illusion of serialised access to the database. Multiple transactions can execute in parallel, so long as they do not interfere by accessing the same data. Where transactions do interfere, some real serialisation can occur. In practice, applications depend on the serialisation and atomicity guarantees given by transactions, often in ways not fully made explicit or understood by the application designers.

Concurrency control in independent, asynchronously replicated systems

Asynchronously replicating writes between two independent systems erodes the guarantees given by single system concurrency control. Each system maintains its transactional guarantees in parallel, and incorporates modifications from the other system asynchronously, at some time after they were originally committed. Where the same row is modified on both systems concurrently, two versions of the same row are produced, and there is no longer a single history of values for the given row. This can cause replicas to diverge. Note that the window of 'concurrency', or 'potential conflict' is related to the time taken for a committed update to be applied on all replicas. This is similar, or equivalent to the commit delay experienced by a synchronous 2-phase commit system.

Conflicts can be detected using some form of conflict detection. On detecting a conflict, steps can then be taken to avoid divergence, and resolve any unwanted effects of the concurrent writes.

Replica divergence, external effects and cascading impacts

Divergence can be avoided if conflicting writes can be merged in some way. Some write conflicts may be equivalent, associative or otherwise mergeable, especially if the operations are replicated rather than their resulting states. However merging requires specific schema and application knowledge to determine how to merge conflicting writes.

More generally, divergence can be avoided by rejecting one or both conflicting writes. This is the approach we have taken, with handling of rejected writes delegated to the application, where the knowledge exists to handle them via the exceptions table mechanism.

However write conflicts are handled, it is important to consider :

1. Cascading impacts on dependent operations
   Operations based on the results of conflicting operations may themselves require handling to avoid divergence.
2. Real world / other system effects based on conflicting writes
   Maintaining database consistency does not guarantee that real world effects have been correctly compensated.
   

A database system does not exist in a vacuum. Operations are performed to reflect external world, or external system events. When the effects of an operation are later reverted, the real world effects may also require some compensating actions. These external world compensating actions are beyond the scope of any DBMS system and are application specific. In a real application of this technology, this is probably the most important part of the design.

Any particular conflict originates between two concurrent operations, but once a conflicting operation is committed, other operations can read its results, and commit their own, expanding the impact of the original conflict. Conflicts are discovered asynchronously, some time after the original operations are committed, so there can be a large number of subsequent operations in the replication pipeline which depend on the conflicting operations at the point they are discovered. All of the invalidated subsequent operations must be handled.

Row based conflict detection and data shearing

Using row-based conflict detection and re-alignment can counteract data divergence so that rows become consistent eventually, but this comes at the cost of eroding the atomicity of committed transactions. For example, a committed transaction which writes to three rows may, after conflict handling, have none, one, two or all three row changes reverted.

Within a single system, the two potentially visible states were :

1. Before transaction (All rows at version 1 : Av1, Bv1, Cv1)
2. After transaction (All rows at version 2 : Av2, Bv2, Cv2)
   

With row based conflict detection, ignoring the row variants we're actually conflicting with, there could be :

Before transaction : (All rows at version 1 : Av1, Bv1, Cv1)

After transaction

1. Av2, Bv2, Cv2 (All rows at version 2)
2. Av2, Bv2, Cv1 (Cv2 reverted)
3. Av2, Bv1, Cv2 (Bv2 reverted)
4. Av2, Bv1, Cv1 (Bv2, Cv2 reverted)
5. Av1, Bv2, Cv2 (Av2 reverted)
6. Av1, Bv2, Cv1 (Av2, Cv2 reverted)
7. Av1, Bv1, Cv2 (Av2, Bv2 reverted)
8. Av1, Bv1, Cv1 (Av2, Bv2, Cv2 reverted)
   

Depending on the concept that the distinct rows A,B,C represented, this can vastly increase the complexity of understanding the data. If A, B and C model entirely separate entities, which just happened to be transactionally updated together then there may be no problem if they fare differently in conflict detection. If they model portions of the state of a larger entity then reasoning about the state of that entity becomes complex.

This potential chopping up of changes committed in a transaction can be described as shearing of the data model represented by the schema. In practice, the potential for shearing between rows implies that for tables with conflicts handled on a row basis, cross row consistency is not available. This in turn implies that the schema must be modified to ensure that data items which cannot tolerate relative shear are placed in the same row so that they share the same fate and remain self-consistent. This single-row limit to consistency is native and natural to some NoSQL / key-value / wide column store products, but is a weakening of the normal guarantees in a transactional system.

Requiring that schemas and applications using conflict detection can tolerate shear between any two rows is quite a heavy burden to place on applications, especially those not written with eventual consistency in mind. Is there some way to support optimistic conflict detection without breaking up committed transactions, and shearing rows?

Transaction based conflict detection

One way to avoid inter-row shearing is to perform conflict detection on a row-by-row basis, but on discovering a conflict, take action on a transaction basis. More concretely, when a row conflict is discovered, any other rows written as part of the same transaction should also be considered in-conflict by implication. This reduces the set of stable states back to the original case - all rows at version 1 or all rows at version 2.

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

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

From our database centric point of view, transactions can only become dependent on each other through the data they access in the database. If transaction X updates rows A and B, and transaction Y then reads row B and updates row C, then we can say that transaction B has a read-write dependency on transaction A via row B. We cannot tell whether there is some other out-of-band communication between transactions.

By tracking this transaction 'footprint' information, and looking for row overlaps, we can determine transaction dependencies. This is how the new NDB$EPOCH_TRANS function provides transactional conflict detection.

NDB$EPOCH_TRANS conflict detection function

The NDB$EPOCH_TRANS conflict function uses the same mechanism as the NDB$EPOCH function to detect concurrent updates to the same row across two clusters. However, once a row conflict has been detected in an operation which is part of a replicated transaction, all other operations in that replicated transaction are considered to be in conflict. Furthermore, any transactions found to be dependent on that transaction are also considered in conflict. Once the full set of in conflict transactions has been determined, the set of affected rows are handled in the same way as in NDB$EPOCH.

Specifically :

• The replicated operations are not applied
• The exceptions table(s) are populated with the affected primary keys
• The affected row epochs are updated
  
• Realignment Binlog events are generated to (eventually) realign the Secondary cluster
  

As with NDB$EPOCH, NDB$EPOCH_TRANS is asymmetric, so the Primary Cluster always wins when a conflict is detected. As with NDB$EPOCH, this allows applications needing pessimistic properties to obtain them by accessing the Primary Cluster. Applications which can handle the relaxed consistency of optimism can access either Cluster. With NDB$EPOCH_TRANS, transactions committed on the Secondary Cluster are guaranteed to be atomic, whether or not they are later found to be in conflict. Each committed transaction will either be unaffected by conflict detection, or be completely reverted. There will be no row shear.

This slightly stronger optimistic consistency guarantee may ease the implementation of relaxed consistency / eventually consistent applications. For example, where some concept is modelled by a number of different rows in different tables, any transactional modification will either be atomically applied, or not applied at all, so the relationships between the rows affected by a transaction will preserved. The need to flatten a schema into single-row entities is reduced, although careful design is still required to get a good understanding of transaction boundaries, and the behaviour of the overall system when transactions are reverted.

Transaction dependency tracking

NDB$EPOCH_TRANS is built in to the MySQL Cluster Storage Engine. It is active in the normal MySQL Slave SQL thread, as part of the normal table handler calls made when applying a replicated Binlog. The NDB$EPOCH_TRANS code in the Ndb storage engine tracks transaction dependencies based on the primary keys accessed by row events in the Binlog, and their transaction ids. If two row events have the same table and primary key values, then they affect the same row. If two events affect the same row, and are in different transactions, then the second transaction depends on the first. In this way, a transaction dependency graph is built by the MySQL Cluster Storage Engine as row events are applied by the Slave from a replicated Binlog. This graph is then used to find dependencies when a conflict is detected.

A Binlog only contains WRITE_ROW, UPDATE_ROW and DELETE_ROW events. This means that we only detect dependencies between transactions which write the same rows. We do not currently track dependencies between writers and readers. For example :

Transaction A : {Write row X, Write row Y}
Transaction B : {Read row Y, Write row Z}

Binlog : {{Tx A : Wr X, Wr Y}, {Tx B : Wr Z}}

In this example, the dependency of Transaction B on Transaction A is not recorded in the Binlog, and so the Slave is not aware of it. This would result in the write to row Z not being considered in conflict, when it should be.

A future improvement is to add selective tracking of reads to the Binlog, so that Write -> Read dependencies will implicate reading transactions when a conflict is discovered.

There's more to come

Another long dry post, best consumed with your favourite drink in hand. As I mentioned last time, these functions are pushed, and available in the latest releases of MySQL Cluster. I'd be happy to hear from anyone who wants to try them out and give feedback. I've been deliberately light with implementation details thus far, as I'm saving those for yet another posting. I think that some of the implementation details are interesting from a replication point of view, even if you're not interested in these particular conflict detection algorithms. You may disagree :)

Edit 23/12/11 : Added index

Eventual consistency with MySQL

tl;dr : New 'automatic' optimistic conflict detection functions available giving the best of both optimistic and pessimistic replication on the same data

MySQL replication supports a number of topologies, and one of the most interesting is an active-active, or master-master topology, where two or more Servers accept read and write traffic, with asynchronous replication between them.

This topology has a number of attractions, including :

• Potentially higher availability
• Potentially low impact on read/write latency
• Service availability insensitive to replication failures
• Conceptually simple

However, data consistency is hard to maintain in this environment. Data, and access to it, must usually be partitioned or otherwise controlled, so that the consistency of reads is acceptable, and to avoid lost writes or badly merged concurrent writes. Implementing a distributed data access partitioning scheme which can safely handle communication failures is not simple.

Relaxed read consistency

Relaxed read consistency is a fairly well understood concept, with many Master-Slave topologies deployed where some read traffic is routed to the Slave to offload the Master and get 'read scaling'.
Generally this is acceptable as :

1. A Read-only Slave's state is self-consistent. It is a state which, at least logically, existed on the Master at some time in the past.
2. The reading application can tolerate some level of read-staleness w.r.t. the most recently committed writes to the Master

A surprisingly large number of applications can manage with a stale view as long as it is self-consistent.

Read-your-writes

Applications requiring 'read your writes' consistency (or session consistency) must either read from the Master, or wait until the Slave has replicated up to at least the point in time where the application's last write committed on the Master before reading from it. It is simpler and less delay-prone to just read from the Master, but this increases the load on the Master, reducing the ability of a system to read-scale. When the Master is unavailable, some sort of failover is required, and therefore, some sort of recovery process is also required.

Partitioned Active-Active/ Balanced Master-Slave

Rather than treating a whole replica as either Master or Slave, we can have each replica be both a Master and a Slave. The partitioning could be on database level, table level, or some function of the rows contained in tables, perhaps some key prefix which maps to application entities. Balancing the Master/Slave role in this way allows the request load to be balanced, reducing issues with a single system providing the Master 'role' having to do more work.

In this configuration, rather than talking about Master and Slave, it makes more sense to talk about some partition of data being 'Active' on one replica, and 'Backup' on the others. Read requests routed to the Active replica will be guaranteed to get the latest state, whereas the Backup replicas can potentially return stale states. Write requests should always be routed to the Active replica to avoid potential races between concurrent writes.

Implementing a partitioned replicated system like this generally requires application knowledge to choose a partitioning scheme where expected transaction footprints align with the partitioning scheme, and cross-partition transactions are rare/non-existant. Additionally, it requires application modification, or a front-end routing mechanism to ensure that requests are correctly routed. The routing system must also be designed to re-route in cases of communication or system failure, to ensure availability, and avoid data divergence. After a failure, recovery must take care to ensure replicas are resynchronised before restoring Active status to partitions in a recovered replica.

Implementing a partitioned replicated system with request routing, failover and recovery is a complex undertaking. Additionally, it can be considered a pessimistic system. For embarassingly parallel applications, with constrained behaviours, most transactions are non-overlapping in their data footprint in space and (reasonable lengths of) time. Enforced routing of requests to a primary replica adds cost and complexity that is most often unnecessary. Is it possible to take a more optimistic approach?

Optimistic Active-Active replication

An optimistic active-active replication system assumes that conflicting operations are rare, and prefers to handle conflicts after they happen, than to make conflicts impossible, by mapping them to delays or overheads all of the time. The one-time cost of recovering from a conflict after it happens may be higher than the one-time cost of preventing a conflict, but this can be a win if conflicts are rare enough.

Practically, optimistic active-active replication involves allowing transactions to execute and commit at all replicas, and asynchronously propagating their effects between replicas. When applying replicated changes to a replica, checks are made to determine whether any conflicts have occurred.

Benefits of optimism include :

• Local reads - low latency, higher availability
• Local writes - low latency, higher availability
• No need to route requests, failover, recover
  Recovery from network failure is the same as for normal async replication - catch up the backlog.

A pessimist is never disappointed, as they always expect the worst, but an optimist is occasionally (often?) disappointed. With active-active replication, this disappointment can include reading stale data, as with relaxed read consistency, or having committed writes later rejected due to a conflict. This is the price of optimism. Not all applications are suited to the slings and arrows inherent in optimism. Some prefer the safety of a pessimistic outlook.

Benefits of pessimism include :

• Only durable data returned by reads
• Committed writes are durable

MySQL Cluster replication has supported symmetric optimistic conflict detection functions since the 6.3 release. These provide detection of conflicts for optimistic active-active replication, allowing data to be written on any cluster, and write-write conflicts to be detected for handling. The functions use an application defined comparison value to determine when a conflict has occurred, and optionally, which change should 'win'. This is very flexible, but can be difficult to understand, and requires application and schema changes to be made use of.

When presented with an either-or decision, why not ask for both? Is it possible to have the benefits of both optimistic and pessimistic replication? Can we have them both on the same data at the same time?


Asymmetric optimistic Active-Active replication

I have recently been working on new asymmetric conflict detection functions for MySQL Cluster replication. These functions do not require schema or application modifications. They are asymmetric in that one data replica is regarded as the Active replica. However, unlike a pessimistic partitioned replicated system, writes can be made at Active or Backup replicas - they do not have to be routed to the Active replica. Writes made at the Backup replica will asynchronously propagate to the Active replica and be applied, but only if they do not conflict with writes made concurrently at the Active replica.

Having a first class Active replica and a second class Backup replica may seem like a weakness. However, it allows optimistic and pessimistic replication to be mixed, on the same data for different use-cases.

Where a pessimistic approach is required, requests can be routed to the Active replica. At the Active replica, they will be guaranteed to read durable data, and once committed, writes will not be rejected later.

Where an optimistic approach is acceptable, requests can also be routed to the Backup replica. At the Backup replica, committed writes may later be rejected, and reads may return data which will later be rejected. The potential for disappointment is there, and applications must be able to cope with that, but in return, they can read and write locally, with latency and availability independent of network conditions between replicas.

A well understood application and schema can use pessimistic replication, with request routing, where appropriate, and write-anywhere active-active where the application and schema can cope with the relaxed consistency.

New conflict functions - NDB$EPOCH, NDB$EPOCH_TRANS

The new NDB$EPOCH function implements asymmetic conflict detection, on a row basis. One replica of a table is considered Active (or Primary), and the other(s) are Backup (or Secondary). Writes originating from the Backup replica are checked at the Active replica to ensure that they don't conflict with concurrent writes originating at the Active replica. If they do conflict, then they are rejected, and the Backup is realigned to the Active replica's state. In this way, data divergence is avoided, and the replicated system eventually becomes consistent.

The conflict detection, and realignment to give eventual consistency all occur asynchronously as part of the normal MySQL replication mechanisms.

As with the existing conflict detection functions, an exceptions table can be defined which will be populated with the primary keys of rows which have experienced a conflict. This can be used to take application specific actions when a conflict is detected.

Unlike the existing conflict detection functions, no schema changes or application changes are required. However, as with any optimistic replication system, applications must be able to cope with the relaxed consistency on offer. Applications which cannot cope, can still access the data, but should route their requests to Active replicas only, as with a more traditional pessimistic system.

As these functions build on the existing MySQL Cluster asynchronous replication, the existing features are all available :

• Slave batching performance optimisations
  
• High availability - redundant replication channels
  
• Transactional replication and progress tracking
  
• Normal MySQL replication features : DDL replication, Binlog, replicate to other engines etc..

Ok, that's long enough for one post - I'll describe NDB$EPOCH_TRANS and its motivations in a follow-up. If you're interested in trying this out, then download the latest versions of MySQL Cluster. If you're interested in the optimistic replication concept in general, I recommend reading Saito and Shapiro's survey.

Edit 23/12/11 : Added index

Some MySQL projects I think are cool - HandlerSocket Plugin

The HandlerSocket project is described in Yoshinori Matsunobu's blog entry under the title 'Using MySQL as a NoSQL - A story for exceeding 750,000 qps on a commodity server'. It's a great headline and has generated a lot of buzz. Quite a few early commentators were a little confused about what it was - a new NoSQL system using InnoDB? A cache? In memory only? Where does Memcached come in? Does it support the Memcached protocol? If not, why not? Why is it called HandlerSocket?

Inspirations from Memcache may include the focus on simplicity, performance and a simple human readable protocol. As Yoshinori says, Kazuho Oku has already implemented a MySQLD-embedded Memcached server, no need to do it again. What's more, the Memcache protocol offers key-value functionality, whereas implementing a new protocol allows more functionality to be exposed.

The choice of name has come in for some flak. I believe the etymology is that HandlerSocket exposes the existing MySQL Handler interface directly over a separate socket. Perhaps a more exciting name will appear at some point, but looking at the MySQL Handler documentation gives a good background on the basis of the HandlerSocket Api.

HandlerSocket implements more than a Key-Value Api. It supports indexed data access in general including :

• Equality search on any index prefix (returning 0, 1 or more rows)
• Inequality search on any index prefix (returning 0, 1 or more rows)

This allows far more general use than a simple key-value API. Composite keys can be used and Secondary indexes searched and maintained. This exposes much more of the value of a storage engine like InnoDB.

SQL and non-SQL access to the same data

Yoshinori mentions inspiration from NdbApi for HandlerSocket - they wanted to get fast indexed access performance without excluding the possibility of performing ad-hoc SQL for reports etc. This has been one of the unique benefits of Ndb for some time - the ability to operate on the same underlying data via multiple Apis. Extending this to other MySQL engines (especially InnoDB) is a great idea. What is surprising is the difference a different access layer implementation can make to throughput. Who would have thought that so much performance could be consumed by parsing etc? I suspect that HandlerSocket may create a new benchmark for the MySQL team to optimise parsing towards in future.

As a MySQL daemon plugin, the HandlerSocket plugin gets to create threads running within a MySQLD server instance. These threads can then listen on network sockets and use the Storage Engine Api inside the server to perform primitive operations on Storage engines. From the point of view of the Storage engine, they are just client request handling threads in the Server accessing data concurrently with 'normal' SQL client threads.

Using the Storage Engine Api, HandlerSocket gets :

• Storage engine independence
• Concurrency control as implemented by the engine
• Index maintenance as implemented by the engine
• Constraint enforcement as implemented by the engine
• Engine features such as online backup, crash recovery, compression, encryption etc.
  

Not going via the 'SQL layer' means that HandlerSocket misses out on :

• SQL functionality (queries, joins, aggregation, UDFs etc.)
  
• Stored procedures
  
• Trigger activation
• Query cache
• ACL checks
  
• Views

As with NdbApi, this is often a very good trade-off as many applications don't need these features for their heavy-lifting. However they can be very useful for less frequent reporting and administration tasks. Supporting consistent access via SQL or some simpler Api to the same data can avoid the need to split caching from an OLTP database and also potentially the need to split an OLTP and analytic database. Ndb gives more of a 'firewall' between different usage types by physically separating the storage from more complex query processing, but MySQLD could be extended to have more 'workload management' features internally if this were a problem for HandlerSocket.

The impressive published benchmarks use data which fits entirely in memory buffers, so that InnoDb need only write logs and checkpoints. HandlerSocket does not require that all data is in-cache, but the best performance will be achieved if this is the case.

Operation batching/pipelining

As the protocol description states, requests can be pipelined. There is no need for individual clients to make blocking synchronous DB requests. This can vastly increase throughput without excessively straining connection resources.

Often SQL DB access protocols do not make much use of the potential for batching requests on a single connection. JDBC supports batched updates and inserts, and appears to have support (not often talked about) for batched queries, but not every driver implements these and they're often not implemented in the same way. Often the SQL approach seems to be 'If you want different sets of data in one round trip, you need to find a way to get them all in one SQL statement'. This creates a false tension between reducing client-server round trips and avoiding complex joins and unnecessary unions. Decoupling query boundaries from thread blocking / flow of control changes removes this artificial tension and can simplify applications. A recent 'Facebook at Scale' talk describes how they use the MySQL Client's CLIENT_MULTI_STATEMENTS flag to decouple query boundaries and request latency. This pattern is one of the keys to implementing efficient NdbApi clients as well.

Commit grouping

Another HandlerSocket feature possibly inspired by Ndb is the commit grouping. Multiple user writes (it's not clear if/how transaction boundaries can be specified), are combined and committed to the engine together. This amortizes the commit cost across multiple operations. Where the engine performs expensive durability operations (e.g fsync) this can improve write throughput. Writes are group-committed to the Binlog as well, again in a similar way to Ndb. Another shared advantage of merging multiple client 'transactions' into fewer Binlog 'transactions' is that the Slave also gets to benefit from fewer, larger transactions for a given data change rate.

Impact

It's not clear yet what the impact of HandlerSocket will be. It enters a crowded market of SQL, NoSQL, NewSQL, DataGrid and other technologies. Competitors use JSON over HTTP, or Memcached protocols, support richer or simpler Apis, offer transparent sharding, eventually consistent replication or web-scale Erlang distributed Map Reduce. Perhaps HandlerSocket is too old-school?

I think it's a great technology, deserving of success. Coupled with an external sharding layer, it seems to offer a great way to improve the efficiency of MySQL scale out, without losing the ability to perform ad-hoc SQL etc. Time will tell.

Journey upriver to the dark heart of ha_ndbcluster

Unlike most other MySQL storage engines, Ndb does not perform all of its work in the MySQLD process. The Ndb table handler maps Storage Engine Api calls onto NdbApi calls, which eventually result in communication with data nodes. In terms of layers, we have SQL -> Handler Api -> NdbApi -> Communication. At each of these layer boundaries, the mapping between operations at the upper layer to operations at the lower layer is non trivial, based on runtime state, statistics, optimisations etc.

The MySQL status variables can be used to understand the behaviour of the MySQL Server in terms of user commands processed, and also how these map to some of the Storage Engine Handler Api calls.

Status variables tracking user commands start with 'Com_'

mysql> show status like 'Com\_%';
+---------------------------+-------+
| Variable_name | Value |
+---------------------------+-------+
| Com_admin_commands | 0 |
| Com_assign_to_keycache | 0 |
| Com_alter_db | 0 |
| Com_alter_db_upgrade | 0 |
| Com_alter_event | 0 |
| Com_alter_function | 0 |
| Com_alter_procedure | 0 |
| Com_alter_server | 0 |
| Com_alter_table | 0 |
| Com_alter_tablespace | 0 |
| Com_analyze | 0 |
| Com_backup_table | 0 |
| Com_begin | 0 |
| Com_binlog | 0 |
| Com_call_procedure | 0 |
| Com_change_db | 1 |
| Com_change_master | 0 |
| Com_check | 0 |
| Com_checksum | 0 |
.........
| Com_stmt_reset | 0 |
| Com_stmt_send_long_data | 0 |
| Com_truncate | 0 |
| Com_uninstall_plugin | 0 |
| Com_unlock_tables | 0 |
| Com_update | 1 |
| Com_update_multi | 0 |
| Com_xa_commit | 0 |
| Com_xa_end | 0 |
| Com_xa_prepare | 0 |
| Com_xa_recover | 0 |
| Com_xa_rollback | 0 |
| Com_xa_start | 0 |
+---------------------------+-------+
144 rows in set (0.01 sec)

mysql>

Status variables tracking Handler (Storage engine) Api calls start with 'Handler_'.

mysql> show status like 'Handler\_%';
+----------------------------+-------+
| Variable_name | Value |
+----------------------------+-------+
| Handler_commit | 1 |
| Handler_delete | 0 |
| Handler_discover | 0 |
| Handler_prepare | 0 |
| Handler_read_first | 0 |
| Handler_read_key | 0 |
| Handler_read_next | 0 |
| Handler_read_prev | 0 |
| Handler_read_rnd | 0 |
| Handler_read_rnd_next | 21 |
| Handler_rollback | 0 |
| Handler_savepoint | 0 |
| Handler_savepoint_rollback | 0 |
| Handler_update | 4 |
| Handler_write | 14 |
+----------------------------+-------+
15 rows in set (0.00 sec)

mysql>

The 'Com_%' and 'Handler_%' variables are maintained by the Server for all storage engines. The server maintains these on a per-session, and global basis. By default the session status is shown, but the GLOBAL keyword shows the global view, aggregated across all sessions.

mysql> show global status like 'Handler\_%';
+----------------------------+--------+
| Variable_name | Value |
+----------------------------+--------+
| Handler_commit | 167 |
| Handler_delete | 494041 |
| Handler_discover | 0 |
| Handler_prepare | 0 |
| Handler_read_first | 3 |
| Handler_read_key | 1 |
| Handler_read_next | 0 |
| Handler_read_prev | 0 |
| Handler_read_rnd | 0 |
| Handler_read_rnd_next | 561132 |
| Handler_rollback | 6 |
| Handler_savepoint | 0 |
| Handler_savepoint_rollback | 0 |
| Handler_update | 24 |
| Handler_write | 43442 |
+----------------------------+--------+
15 rows in set (0.00 sec)

mysql>

The SHOW STATUS command is handy for a quick check, but for more interesting analysis, the INFORMATION_SCHEMA tables SESSION_STATUS and GLOBAL_STATUS contain the same data, and support all SQL queries and views.

mysql> select * from information_schema.session_status where Variable_name like 'Handler\_%';
+----------------------------+----------------+
| VARIABLE_NAME | VARIABLE_VALUE |
+----------------------------+----------------+
| HANDLER_COMMIT | 1 |
| HANDLER_DELETE | 0 |
| HANDLER_DISCOVER | 0 |
| HANDLER_PREPARE | 0 |
| HANDLER_READ_FIRST | 0 |
| HANDLER_READ_KEY | 0 |
| HANDLER_READ_NEXT | 0 |
| HANDLER_READ_PREV | 0 |
| HANDLER_READ_RND | 0 |
| HANDLER_READ_RND_NEXT | 85 |
| HANDLER_ROLLBACK | 0 |
| HANDLER_SAVEPOINT | 0 |
| HANDLER_SAVEPOINT_ROLLBACK | 0 |
| HANDLER_UPDATE | 4 |
| HANDLER_WRITE | 89 |
+----------------------------+----------------+
15 rows in set (0.40 sec)

mysql>

Unfortunately the entries in the information_schema are in shouty 1960s CAPITALS. Also note that some Handler calls are made as part of using the information_schema 'database' to fetch the data. This is Heisenberg's principle in action!

To shine some light into the depths of the Ndb storage engine, a set of new status variables has been added to recent cluster-7.0 and cluster-7.1 releases. These status variables track activity at the NdbApi and data node communication layers of the stack. Currently they are divided into 4 subsets :

• Global counters
  show status like 'ndb_api_%_count';
• Session counters
  show status like 'ndb_api_%_session';
• Slave counters
  show status like 'ndb_api_%_slave';
• Binlog injector counters
  show status like 'ndb_api_%_injector';
  

Unfortunately the mysql-5.1 server does not allow Storage Engines to differentiate GLOBAL or SESSION status variables, so the Global and Session specific versions of these variables are differentiated by name, and are visible from both GLOBAL and SESSION views of status, so it doesn't matter which you look at.

Global counters
mysql> show status like 'ndb_api_%_count';
+------------------------------------+------------+
| Variable_name | Value |
+------------------------------------+------------+
| Ndb_api_wait_exec_complete_count | 10 |
| Ndb_api_wait_scan_result_count | 19295 |
| Ndb_api_wait_meta_request_count | 67 |
| Ndb_api_wait_nanos_count | 6361966340 |
| Ndb_api_bytes_sent_count | 415704 |
| Ndb_api_bytes_received_count | 116921552 |
| Ndb_api_trans_start_count | 14 |
| Ndb_api_trans_commit_count | 3 |
| Ndb_api_trans_abort_count | 1 |
| Ndb_api_trans_close_count | 14 |
| Ndb_api_pk_op_count | 6 |
| Ndb_api_uk_op_count | 0 |
| Ndb_api_table_scan_count | 11 |
| Ndb_api_range_scan_count | 0 |
| Ndb_api_pruned_scan_count | 0 |
| Ndb_api_scan_batch_count | 25850 |
| Ndb_api_read_row_count | 103371 |
| Ndb_api_trans_local_read_row_count | 51625 |
| Ndb_api_event_data_count | 0 |
| Ndb_api_event_nondata_count | 0 |
| Ndb_api_event_bytes_count | 0 |
+------------------------------------+------------+
21 rows in set (0.00 sec)

mysql>

These counts are aggregated across all MySQL clients in the Server accessing tables in Ndb, since this MySQL Server was started. Note that this *does not* include accesses by clients of other MySQL Servers, or other NdbApi clients.

Session counters
mysql> show status like 'ndb_api_%_session';
+--------------------------------------------+----------+
| Variable_name | Value |
+--------------------------------------------+----------+
| Ndb_api_wait_exec_complete_count_session | 0 |
| Ndb_api_wait_scan_result_count_session | 38 |
| Ndb_api_wait_meta_request_count_session | 2 |
| Ndb_api_wait_nanos_count_session | 11064398 |
| Ndb_api_bytes_sent_count_session | 872 |
| Ndb_api_bytes_received_count_session | 230764 |
| Ndb_api_trans_start_count_session | 1 |
| Ndb_api_trans_commit_count_session | 0 |
| Ndb_api_trans_abort_count_session | 0 |
| Ndb_api_trans_close_count_session | 1 |
| Ndb_api_pk_op_count_session | 0 |
| Ndb_api_uk_op_count_session | 0 |
| Ndb_api_table_scan_count_session | 1 |
| Ndb_api_range_scan_count_session | 0 |
| Ndb_api_pruned_scan_count_session | 0 |
| Ndb_api_scan_batch_count_session | 51 |
| Ndb_api_read_row_count_session | 204 |
| Ndb_api_trans_local_read_row_count_session | 136 |
+--------------------------------------------+----------+
18 rows in set (0.00 sec)

mysql>

These counts are for the current session (MySQL client connection)

Slave counters
mysql> show status like 'ndb_api_%_slave';
+------------------------------------------+-------+
| Variable_name | Value |
+------------------------------------------+-------+
| Ndb_api_wait_exec_complete_count_slave | 0 |
| Ndb_api_wait_scan_result_count_slave | 0 |
| Ndb_api_wait_meta_request_count_slave | 0 |
| Ndb_api_wait_nanos_count_slave | 0 |
| Ndb_api_bytes_sent_count_slave | 0 |
| Ndb_api_bytes_received_count_slave | 0 |
| Ndb_api_trans_start_count_slave | 0 |
| Ndb_api_trans_commit_count_slave | 0 |
| Ndb_api_trans_abort_count_slave | 0 |
| Ndb_api_trans_close_count_slave | 0 |
| Ndb_api_pk_op_count_slave | 0 |
| Ndb_api_uk_op_count_slave | 0 |
| Ndb_api_table_scan_count_slave | 0 |
| Ndb_api_range_scan_count_slave | 0 |
| Ndb_api_pruned_scan_count_slave | 0 |
| Ndb_api_scan_batch_count_slave | 0 |
| Ndb_api_read_row_count_slave | 0 |
| Ndb_api_trans_local_read_row_count_slave | 0 |
+------------------------------------------+-------+
18 rows in set (0.00 sec)

mysql>

Hopefully you are seeing a pattern here. These counters are the NdbApi operations performed by the Slave SQL thread as part of replicating Binlogs into Ndb tables. These counts will only increase from zero if the MySQLD is acting, or has acted as a Slave, and has accessed tables stored in Ndb.

Binlog Injector counters
mysql> show status like 'ndb_api_%_injector';
+--------------------------------------+-------+
| Variable_name | Value |
+--------------------------------------+-------+
| Ndb_api_event_data_count_injector | 0 |
| Ndb_api_event_nondata_count_injector | 0 |
| Ndb_api_event_bytes_count_injector | 0 |
+--------------------------------------+-------+
3 rows in set (0.01 sec)

mysql>

These counts track the data change events received by the Ndb Binlog Injector thread. The Binlog Injector is responsible for recording Cluster changes in the Binlog, but even when Binlogs are not being written, it receives events related to schema changes and other system management functions, so these counts can be non zero on Servers which are not writing a Binlog.

Counter definitions
As you've hopefully noticed, there is naming overlap between each set of status counters. The same events are being counted, and recorded globally, per-session, against the Slave SQL thread and against the Ndb Binlog injector thread.

So what do these different counts actually mean?

Ndb_api_wait_exec_complete_count[_session|_slave]

The number of times a user thread has blocked waiting for some batch of primary key, or secondary unique hash key operations to complete. From the point of view of the MySQL Server, this is idle time, waiting for data nodes to send some response. An alternative name for it could be 'round trip count', and minimising it through operation batching is a good way to reduce response time and increase throughput.

Ndb_api_wait_scan_result_count[_session|_slave]

The number of times a user thread has blocked waiting for some scan operation to complete. It could be waiting for a batch of scan results, or waiting for an acknowledgement of a scan close. In any case, it indicates time spent waiting on communication related to scan processing.

Ndb_api_wait_meta_request_count[_session|_slave]

The number of times a user thread has blocked waiting for some metadata operation to complete. This is quite a catch-all term, which can include DDL (Create/Drop table etc), and some transaction initialisation steps.

Ndb_api_wait_nanos_count[_session|_slave]

The number of nanoseconds a user thread has blocked in one of the three scenarios above. This is kind of an 'IO_WAIT' time for Ndb operations. It tracks how long the thread was blocked waiting for the data nodes to complete their operations. The resolution is nanoseconds, but this requires support from the operating system. On operating systems with lower resolution, this count will be coarser, and some operations may complete with zero observed wait time.

Ndb_api_bytes_sent_count[_session|_slave]

The number of bytes sent to the Ndb data nodes. This includes all request types, rows inserted etc. It does not include regular heartbeating as that generally adds too much noise to make the counters useful.

Ndb_api_bytes_received_count[_session|_slave]

The number of bytes received from the Ndb data nodes. This includes all request types, rows read etc. It does not include regular heartbeating as that generally adds too much noise to make the counters useful.

Ndb_api_trans_start_count[_session|_slave]

The number of NdbApi transactions started. Note that NdbApi transactions are not always immediately on a BEGIN statement as an optimisation.

Ndb_api_trans_commit_count[_session|_slave]

The NdbApi transactions which have been explicitly committed. Not all transactions started are committed or aborted, some are started, and then closed.

Ndb_api_trans_abort_count[_session|_slave]

The NdbApi transactions which have been explicitly aborted.

Ndb_api_trans_close_count[_session|_slave]

The NdbApi transactions which have been closed. It should closely track the number which have been started.

Ndb_api_pk_op_count[_session|_slave]

The number of Primary Key (pk) operations which have been executed. Eack pk operation affects zero or one rows. This includes read, insert, update, write, delete. Note that operations on tables with Blobs can also generate pk and uk operations.

Ndb_api_uk_op_count[_session|_slave]

The number of Unique key (uk) operations which have been executed. Each uk operation affects zero or one rows. This includes read, update, write, delete by unique key.

Ndb_api_table_scan_count[_session|_slave]

The number of table scans which have been started. Table scans can have pushed-down filters, so although they must access all data in a table, they might not return it all to the MySQL Server. Also, started scans may be stopped before all table data is accessed.

Ndb_api_range_scan_count[_session|_slave]

The number of range (Ordered Index) scans which have been started. Range scans take bounds and pushed-down filters, so may access and/or return zero to all rows to the MySQL Server.

Ndb_api_pruned_scan_count[_session|_slave]

The number of scans which have been successfully pruned to one partition of the scanned table/index. See my previous Blog entries for details about scan partition pruning.

Ndb_api_scan_batch_count[_session|_slave]

The number of batches of rows returned to the MySQL Server from scans. Each scanned table/index fragment returns matching rows in batches, whose size is controlled by the batchsize parameters. NdbApi handles one batch from each fragment at a time. Fetching the next batch from a fragment requires a round-trip to the data nodes, although multiple fragments can be asked for their next batch in one trip. Minimising the Ndb_api_scan_batch_count, by increasing batchsize and improving scan selectivity can improve throughput, latency and efficiency.

Ndb_api_read_row_count[_session|_slave]

The number of rows returned to the MySQL Server from Primary Key read, Unique Key reads and Table and Index scans.

Ndb_api_trans_local_read_row_count[_session|_slave]

The number of rows returned to the MySQL Server from the same data node where the reading transaction has its Transaction Coordinator (TC). Where the TC and the data read reside on the same data node, one hop in the data reading control protocols is avoided. This is the goal of transaction hinting and distribution awareness, and its effectiveness can be checked by comparing Ndb_api_trans_local_read_row_count to Ndb_api_trans_read_row_count. The higher the proportion of local reads, the better.

Ndb_api_event_data_count[_injector]

The number of data change events (row insert, delete, update notifications) received from the data nodes. On a Binlogging MySQL Server, this count can give a measure of the rate of data change in a cluster in terms of rows/second.

Ndb_api_event_nondata_count[_injector]

The number of non-data events (table alter/drop notifications etc) received from the data nodes.

Ndb_api_event_bytes_count[_injector]

The total number of bytes of event data (data and nondata events) received from the data nodes. This gives another measure of the Cluster change rate in terms of bytes/second.

At NdbApi level
As the names suggest, all of these counters are reflecting things happening in the NdbApi implementation. The data collection is built into NdbApi, and is therefore also available to any NdbApi client, not just the MySQL Server.

Using the counters
During implementation of these counters, I found it easiest to create a temporary table to store 'base' values for the counters, and define a view containing the difference between the current values and the base values. This made it easy to see the effect on the counters of various different SQL statements, slave operations etc.

Setting up a baseline and a view
Please excuse my schoolboy SQL :

mysql> create table test.counts_base(variable_name varchar(255) primary key, variable_value bigint);
Query OK, 0 rows affected (0.04 sec)

mysql> replace into test.counts_base (variable_name, variable_value) select * from information_schema.session_status where variable_name like 'ndb_api%';
Query OK, 60 rows affected (0.01 sec)
Records: 60 Duplicates: 0 Warnings: 0

mysql> create view test.counts_diff as select test.counts_base.variable_name, information_schema.session_status.variable_value - test.counts_base.variable_value as diff from test.counts_base, information_schema.session_status where test.counts_base.variable_name = information_schema.session_status.variable_name and (information_schema.session_status.variable_value - test.counts_base.variable_value) > 0;
Query OK, 0 rows affected (0.05 sec)

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
Empty set (0.05 sec)

mysql>


Looking at the effects of a SQL statement
First check that the baseline is ok :

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
Empty set (0.05 sec)

Now run the statement :

mysql> select count(1) from demo_table;
+----------+
| count(1) |
+----------+
| 103343 |
+----------+
1 row in set (0.00 sec)

Now look at the difference :

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
+--------------------------------------------+--------+
| variable_name | diff |
+--------------------------------------------+--------+
| NDB_API_WAIT_SCAN_RESULT_COUNT_SESSION | 4 |
| NDB_API_WAIT_META_REQUEST_COUNT_SESSION | 1 |
| NDB_API_WAIT_NANOS_COUNT_SESSION | 619143 |
| NDB_API_BYTES_SENT_COUNT_SESSION | 116 |
| NDB_API_BYTES_RECEIVED_COUNT_SESSION | 268 |
| NDB_API_TRANS_START_COUNT_SESSION | 1 |
| NDB_API_TRANS_CLOSE_COUNT_SESSION | 1 |
| NDB_API_TABLE_SCAN_COUNT_SESSION | 1 |
| NDB_API_SCAN_BATCH_COUNT_SESSION | 2 |
| NDB_API_READ_ROW_COUNT_SESSION | 2 |
| NDB_API_TRANS_LOCAL_READ_ROW_COUNT_SESSION | 1 |
+--------------------------------------------+--------+
11 rows in set (0.05 sec)

mysql>

The select count(1) statement blocked 4 times on scan results, sent 116 bytes and received 268 bytes of data. Two batches of rows were received, and two rows were received in total. One of these rows was from the same node as the transaction's transaction coordinator.
This indicates that select count(1) is optimised in the Ndb handler !

Let's try a more tricky select count. First we must reset the baseline to get a 'clean' difference.

mysql> replace into test.counts_base (variable_name, variable_value) select * from information_schema.session_status where variable_name like 'ndb_api%';
Query OK, 120 rows affected (0.01 sec)
Records: 60 Duplicates: 60 Warnings: 0

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
Empty set (0.05 sec)

mysql> select count(length(string_value) > 10) from demo_table;
+----------------------------------+
| count(length(string_value) > 10) |
+----------------------------------+
| 103343 |
+----------------------------------+
1 row in set (6.77 sec)

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
+--------------------------------------------+------------+
| variable_name | diff |
+--------------------------------------------+------------+
| NDB_API_WAIT_SCAN_RESULT_COUNT_SESSION | 21139 |
| NDB_API_WAIT_NANOS_COUNT_SESSION | 5322402052 |
| NDB_API_BYTES_SENT_COUNT_SESSION | 441628 |
| NDB_API_BYTES_RECEIVED_COUNT_SESSION | 109026900 |
| NDB_API_TRANS_START_COUNT_SESSION | 1 |
| NDB_API_TRANS_CLOSE_COUNT_SESSION | 1 |
| NDB_API_TABLE_SCAN_COUNT_SESSION | 1 |
| NDB_API_SCAN_BATCH_COUNT_SESSION | 25836 |
| NDB_API_READ_ROW_COUNT_SESSION | 103343 |
| NDB_API_TRANS_LOCAL_READ_ROW_COUNT_SESSION | 51735 |
+--------------------------------------------+------------+
10 rows in set (0.05 sec)

mysql>

This select count waited for scan results ~21 thousand times, sent ~440kB of data, and received ~109MB of data from the data nodes. 103,343 rows were read in ~25 thousand scan batches, and roughly half came from the same node as the tranaction coordinator. 5.3 seconds of the 6.77 second runtime were spent waiting for responses from the data nodes.

Other uses
Hopefully this gives some notion of the possibilities with these new counters. Some other ideas :

• Debug slow queries
• Optimise data distribution and table partitioning
• Get real post-execution costs for queries, DML etc.
• Understand how data transfer for Blobs is batched
• Check bulk insert/deletes are functioning efficiently
• Verify Ndb slave batching is in operation
• Draw cool graphs
  

Low latency distributed parallel joins

When MySQL AB bought Sun Microsystems in 2008 (or did Sun buy MySQL?), most of the MySQL team merged with the existing Database Technology Group (DBTG) within Sun. The DBTG group had been busy working on JavaDB, Postgres and other DB related projects as well as 'High Availability DB' (HADB), which was Sun's name for the database formerly known as Clustra.

Clustra originated as a University research project which spun out into a startup company and was then acquired by Sun around the era of dot-com. A number of technical papers describing aspects of Clustra's design and history can be found online, and it is in many ways similar to Ndb Cluster, not just in their shared Scandinavian roots. Both are shared-nothing parallel databases originally aimed at the Telecoms market, supporting high availability and horizontal scalability. Clustra has an impressive feature set and many years of development behind it, but limited exposure to general purpose use.

At the time of the MySQL acquisition, HADB/Clustra was embedded in a number of Sun products as a session store and metadata repository, but was not available for external customers for general purpose use. Shortly afterwards, a decision was made to move HADB into a 'sustaining' model, and most of the ex-HADB team then became available to work on other projects. MySQL has greatly benefited from the injection of skills and enthusiasm from the Sun DBTG across a number of different teams, which is maybe not well known to those outside the company.

In the Cluster team, one project which has really benefited is the SPJ (Select Project Join) project, which couldn't have happened without the expertise and energy of the ex-Clustra/HADB team working on it.

The SPJ project started around the time of the last MySQL Developers conference in Riga in September 2008. The intention at the time was to look at ways of efficiently supporting more complex queries, specifically involving table joins, reducing unnecessary data transfer, communication latencies and context switches and increasing parallelism.

The main insights at the start of the project included :

• Join mechanisms should be based on linking existing NdbApi single-table access primitives
• Join result sets need not and should not be fully materialised at the data nodes
• Join mechanisms need not be fully general or fully capable initially, as full generality/capability is already available with the existing Apis
• Small joins should be targeted (Number of rows, number of tables)
  

These points have simplified the project scope greatly, and allowed many painful and costly detours to be avoided.

SQL execution in MySQL Cluster

As described in a previous post, Ndb Cluster was originally designed to quickly execute small queries with a high update rate at low latency. Larger more complex queries were executed by a separate query processor. The emphasis in this design is that complex queries are possible, but not necessarily fast or efficient. A main goal is that complex queries do not adversely affect the properties of the high volume, low latency requests.

All data access in MySQL Cluster is via the NdbApi interface. The NdbApi gives access to data stored in tables in the Cluster via four primitives :

• Read row by primary key
• Read row by secondary unique key
• Scan a range of rows in an ordered index with optional conditions
• Scan all rows in a table with optional conditions
  

Each of these primitives operates on a single table, and any table joins must be done by the NdbApi user. Different primary key/unique key operations and individual scan operations run in parallel across the data nodes in a cluster. NdbApi allows different operations and scan requests to be batched together to minimise latency due to communication delays, and it is essential to use this batching to get minimal latencies with Ndb.

In MySQL Cluster, attached MySQL Servers act as query processors, and MySQL's SQL execution engine breaks complex queries down into calls to its generic Storage Engine (SE) Api, which also deals with data access one table at a time.

The Ndb storage engine then further decomposes these SE Api calls into NdbApi primitive operations on individual tables.

MySQL supports SQL queries by performing the SE Api calls to read data, then comparing and matching results, sorting, buffering and reformatting. This works very well and gives MySQL Cluster great SQL functionality and compatibility, although users may find that the latency of their individual queries is not as low as with other engines such as MyISAM and InnoDB, which do not have to perform inter-process communication to implement their SE Api calls.

For minimum latency, Ndb requires that the MySQL Server makes efficient requests for data, requesting as much data as possible at once, and not using the data until it is essential to make forward progress - e.g. when there is a real data dependency.

MySQL features such as Insert, Update and Delete batching, and Batched Key Access minimise the MySQLD to Data node round trips required to execute certain types of operations, but they are unable to help when there are real data dependencies in a query. For example when the server needs to read some value from table t1 to know which rows to read from table t2 then there is no alternative but to read the t1 rows into memory before issuing any reads from t2.

Linked Operations

To reduce the need for extra Api to Data node round trips for every data dependency, we must allow operations to be linked. If we can describe the data dependency as a link between NdbApi operations, then it can be resolved amongst the data nodes. For example, rather than stating :

 SQL > select t1.b, t2.c from t1,t2 where t1.pk=22 and t1.b=t2.pk;
        ndbapi > read column b from t1 where pk = 22;
           

                   [round trip]
           

                   (b = 15)
        ndbapi > read column c from t2 where pk = 15;
          

                   [round trip]
          

                   (c = 30)
        [ return b = 15, c = 30 ]

We would state the join/operation linkage at the ndbapi level :

        ndbapi > read column @b:=b from t1 where pk = 22;
                 read column c from t2 where pk=@b;
          

                   [round trip]
          

                   (b = 15, c = 30)
        [ return b = 15, c = 30 ]

We allow read operations to be parameterised on the results of previous operations, and have the linking of the operations, the flow of results into parameters, handled by the data nodes. The data dependency still results in some execution serialisation at the data node layer, but not at the api layer, so data dependencies within queries needn't result in extra round trips between the MySQL server and the data nodes. Where the dependent data happens to be on the same data node, the dependency can be resolved with no inter-process communication at all.

Viewing the database software as a stack, the execution of the join is being 'pushed down' the stack, to a lower layer. For this reason, the SPJ functionality is also sometimes referred to as pushed-down joins or just pushed joins. Pushing functionality closer to the data can result in improved performance due to lower latency, reduced data transfer etc. In the case of MySQL Cluster, it can avoid inter-process communication, as well as enable parallelism across the data nodes.

In theory, linking can occur between any of the 4 primitive operation types :

• Primary key access (PK)
• Unique key access (UK)
• Ordered index range scan (OI)
  (Range bounds and optional conditions parameterised)
• Table scan (TS)
  (Optional conditions parameterised)
  

In practice, linking the cardinality (0|1) operations (Primary key, Unique key) together is simpler than linking with the scans. In turn, linking a scan to a cardinality (0|1) operation is simpler than linking a scan to another scan.

Linking a table scan to a table scan results in a cross-join and is probably going to be unpleasantly expensive for anything other than small tables.

The initial SPJ implementation supports combinations of Primary/Unique key operations linked together with at most one ordered index scan.

A future implementation will support multiple ordered index scans in a single request. This is more complex to handle due to the buffering required of the different scan result sets, and the resulting result ordering versus efficiency tradeoffs.

The SPJ Api is implemented as an extension to the existing NdbApi, with similar primitive concepts, but with the addition of the means to link the primitives together. As with the existing NdbApi, the usage pattern is along the lines :

• Define operation(s)
• Define further linked operation(s)
• Execute() // One round trip to the data nodes
  
• Examine results
  

In terms of batching, a tree of linked operations, with one const-parameterised root operation, and one or more child operations, is considered to be a single operation. Multiple SPJ operations, each actually a tree of primitive operations, can be executed simultaneously in a batch, along with other 'basic' NdbApi operations.

Where a scan is included, the scan can be advanced using the normal nextResult() mechanism, which also advances the results returned by any cardinality (0|1) child operations.

NoJoins - Not only Joins

While the SPJ extensions are described here in terms of joins, at the NdbApi level they are really 'linked operations'. One design goal which is not completely aligned with the join concept was to allow scans of multiple different tables to be parameterised on a single root operation. For example :

• read @eid:= entity_id from map_table where username="jan";
• scan blog_titles from blog_posts where entity_id=@eid;
• scan latest_tweets from twitter_feed where entitiy_id = @eid;
• scan share_prices from stock_feed where entity_id = @eid;
• ....

Here there is a data dependency between the first lookup and n peer child scans. I want to read all of this data in one round trip, but I don't necessarily want to have to express this in a single 'join' query. If we had a more relational/SQL oriented Api we might have had to create some unholy union of the different results, with masses of repeated values or nulls, or repeat the first lookup for each of n two-way joins.

With the linked operation concept, we can clearly state that the child scans are parameterised by the first lookup, without having to introduce some further unnatural coupling between the rows returned by each scan, which are otherwise independent.

So although SPJ is named after and described as supporting joins, it doesn't mean that you have to be 'join-oriented' or a SQL Samurai to benefit from it. It may be quite useful for efficiently traversing graphs, hierarchies and other links between rows where the concept of a 'join' is quite alien.

Check it out

A mysql-5.1-cluster-7.1 source tree with the SPJ enhancements can be downloaded from here. You can see the NdbApi extensions in the storage/ndb/include/ndbapi directory of the source tree. This source also includes extensions to the MySQL Ndb handler to make use of the new SPJ Api for SQL queries, which I hope to describe a little next time. If you want to download and try out SPJ then see some of the other blog posts about how to get started with it.

Latency hiding patterns in CPUs

Latency is a major factor in CPU design. Most general purpose CPUs execute a sequence of instructions with the logical convention that one instruction completes before the next starts. However memory access latency and even the latency of pure computation bottleneck the achievable processing throughput. To circumvent this a number of techniques are used to parallelise computation and communication.

Conductors have capacitance and resistance and therefore take time to switch between voltage levels. This manifests as a propagation delay proportional to the length of the conductor. As CPU clock speeds increase, the time between each rising clock edge reduces and the maximum length of conductor that can be charged in that time shrinks. CPUs and other data-path designs must be designed to ensure that no signal propagation path is near the tolerance for propagation delay at the designed maximum clock speed.

In practice this means that there is a trade-off between clock frequency and circuit size. Large circuits can accomplish more per clock cycle, but cannot be clocked as high as smaller circuits. To continue increasing clock rates with a fixed feature size, a circuit must be broken up into smaller and smaller sub-circuits with synchronous boundaries. Even this is not enough in some cases as the clock signals themselves are skewed by the propagation latencies across the chip. In some cases this is solved by having islands of synchronous logic which communicate asynchronously as no synchronous clock can span the whole circuit.

So even within a chip, there can be a large, and growing difference between local and remote communication latency. Ignoring this simplifies designs but forces lowest common denominator performance. This trade-off between complexity and latency awareness and tolerance is repeated at many layers of the system stack.

Even within a synchronous logic island on a chip, there is latency to be hidden. Instruction fetch, decode and execute takes time and pipelining used in CPU designs to maximise throughput despite irreducible latency. The various steps of instruction decode, register selection, ALU activation etc. are split out with latches between so that the maximum propagation delay in any stage of the pipleine can be minimised. For example, a datapath with a 400 gate worst-case propagation delay could theoretically be split into 4 stages, each with a 100 gate worst case propagation delay, allowing a 4 times faster clock speed.

With pipelining in a CPU, the first instruction to be processed takes at least the same amount of time to be completed, but after that, a new instruction can be completed on every clock cycle - theoretically allowing 4 times as many instructions to be completed per unit time. In practice memory latency and bandwidth limits and dependencies between instructions mean that the pipeline can stall, or bubbles can form, reducing the achieved throughput. However, these problems can often be alleviated and pipelines have been very successful in CPU design with instruction fetch-execute-retire pipelines comprising as many as 40 or more stages with multiple instances of stages in super-scalar designs

So pipelining in general allows repetitive serial task processing to be parallelised with potential benefits :

• Increased parallelism between instances of serial tasks
  Allowing greater throughput
  
• Benefits of task-specificity (instruction and data caching benefits)
  Potentially improving efficiency
  

and the following caveats :

• Dependencies between tasks must be understood and honoured
  Sometimes this reduces or eliminates parallelism
• Pipeline stalls or bubbles due to job starvation or dependencies will reduce throughput
  The pipeline must be kept fed.
  
• Achievable throughput is bottlenecked by the slowest stage
  As with a production line, the slowest worker sets the pace.
  As with a production line, slow workers can be duplicated to improve throughput.
• Individual task latency can increase
  

Pipelining as a general technique can apply to any system that processes tasks of a regular form which can be split into stages. The SEDA system from Harvard is a general purpose server framework for implementing server processes as pipelined stages. The software setting allows more flexible and dynamic tradeoffs to be made between pipeline lengths and widths. It also offers a flexible way to separate asynchronous and synchronous steps.

Other interesting latency hiding techniques seen in CPUs are mostly associated with hiding memory access latency.

Latency hiding patterns

I want to write an entry about latency hiding patterns. Unfortunately my previous attempts became too long and boring even for me. This time I'm going to try something smaller and less rich to get the blog-flow more regular.

I'm interested in latency hiding patterns as I repeatedly see them being implemented at all levels of systems from the silicon to the top of the application stack. Often latency hiding techniques are what deliver better-than-Moore's law performance improvements and can have great impacts to usability as well as throughput and system efficiency.

I would describe latency hiding techniques as things that :

• Maximise the value of communication
• Maximise the concurrency of communication and computation

Communication value is maximised by avoiding communication, and when it cannot be avoided, minimising the overheads.
Communication and computation concurrency is maximised by maximising the independence of communication and computation. This requires understanding the dependencies between computation and communication.

Right, all very abstract. Let's hope some examples are more interesting.