MySQL Cluster Backup and Restore

MySQL Ndb Cluster provides durability for data by default via logging and checkpointing.

In addition, users can take backups at any time, which allows for disaster recovery, replication synchronisation, data portability and other use cases.

This post looks at the backup and restore mechanisms in MySQL Ndb Cluster.

MySQL Ndb Cluster architecture recap

MySQL Ndb Cluster is a distributed SQL relational database :

• Designed for low overhead read + write scale out, high availability, high throughput and low latency.
• Providing distributed parallel joins, transactions, row locks, foreign keys.
• Data is primarily stored and managed by a set of independent data node processes.
• Data is accessed via distributed MySQL servers and other clients.
• Performance
  
• Scale out read + write : Tables and indices are distributed across data nodes.
• Scale up read + write : Tables and indices are distributed across Local Data Manager (LDM) components within each data node, with parallel checkpointing and redo logs.
• Scale out + up of read + write : Fully parallel and distributed transaction prepare + commit.
• High availability :
• Tables and indices transactionally replicated across data nodes for high availability, locality, read scaling.
• Automatic logging, checkpointing, failure detection, failover, recovery, resynchronisation
• Online operations : Backup, schema changes, upgrade.
• MySQL compatibility, async replication between clusters and InnoDB, connectors, 3rd party support.
  

MySQL Ndb Cluster is a superset of the MySQL Server functionality, and users may use a mix of Ndb and InnoDB tables and features.

Ndb backup + ndb_restore

MySQL Ndb Cluster has a built-in backup mechanism called Ndb backup.  Ndb backup captures the state of all of the tables and indexes stored in the cluster data nodes, allowing them to be restored to a transactionally consistent point in time.  As this data is partitioned and distributed across the cluster data nodes, the backup is also partitioned and distributed, with each data node producing a standalone set of backup filesets containing a subset of the cluster's data.  Restoring all of the backup filesets results in restoring all of the cluster's data.

Ndb backup filesets are restored using a tool called ndb_restore.  This tool can read the filesets generated by Ndb backup, and uses NdbApi to efficiently restore the data to a target cluster.  The backup files contain a logical rather than physical format - restoration is effectively re-insertion and modification of rows.  

This design allows ndb_restore to be flexible about restoring data, supporting :
 - Restore to a different target schema (1, 2, 3, 4, 5, 6)
 - Restore of a subset of tables and indices (1, 2, 3, 4)
 - Restore to a different table or index distribution
 - Restore to a different cluster size or topology
 - Restore to a different cluster software version
 - Restore to a different hardware or OS platform
 - Remote restore across a network

 

As the backup is partitioned into logical shared-nothing filesets, the restore is also naturally partitioned and parallelised for minimal restoration time.

Note that Ndb Backup captures only the data stored in the data nodes, e.g. tables with ENGINE=NDB[CLUSTER].  InnoDB tables and indices, and other MySQL Server local objects such as Triggers, Stored procedures and Views must be backed up and restored using different mechanisms such as mysqldump.

Binary logs

Each Ndb Backup captures the state of a cluster at a transactionally consistent point in time, represented in Ndb as an epoch number.  Epoch numbers are also used in MySQL Ndb Cluster as transactional Binary log markers, dividing the parallel flow of committed transactions into consistent units.  This transactional alignment between backups and binary logs makes it easy to setup and manage async replication between clusters, and implement point in time recovery solutions.  The binlog position corresponding to a backup can be precisely determined, and replication can be established from that point onwards.


Indexes and Constraints

MySQL Ndb Cluster supports secondary unique and ordered indexes, and transactional foreign key constraints.  Unique and foreign key constraints are not enforceable until backup restoration is complete, and maintaining secondary indexes during data restoration consumes resources.  ndb_restore has options to temporarily disable and then rebuild secondary indexes and constraints around the bulk data loading phase of restore allowing :
 - Optimised restoration without the need to maintain indexes, check constraints
 - Avoidance of dependency problems with e.g. circular foreign key references, self references etc
 - Optimised parallel index build

All constraints are naturally checked as part of the constraint rebuild, so integrity is enforced.

Schema changes

Ndb backup filesets capture the schema of tables and indices.  ndb_restore uses this information to recreate tables and indexes prior to restoring data.  

When restoring data, ndb_restore compares the schema of tables and indexes in the target cluster to those captured in the backup.  ndb_restore optionally allows a number of differences, e.g. in numbers of columns, type widths etc, so that data captured from an old schema version can be restored into a new schema without intermediate transformation steps.

Recent versions have added support for column remapping, where data can be transformed, and changes to the set of primary key columns, allowing distribution optimisations.

These features can be used alongside the schema flexibility features of MySQL asynchronous replication, allowing replication to be established and synchronised from an old schema cluster to a new schema cluster, and vice-versa.

Compression

Ndb backup filesets can optionally be compressed using a parameter in the cluster configuration.  Compression is performed by background IO threads in the data node processes, prior to writing data to disk.  ndb_restore handles compressed backups automatically during restoration.

Encryption

From the 8.0.22 release, Ndb backup filesets can be encrypted.  Backup encryption provides protection for data which must leave the cluster security perimeter, and combines with e.g. MySQL Binlog encryption to help implement a secure async replication mechanism.  Backup encryption occurs in background IO threads in the data node processes.

Backups are encrypted and decrypted using a user supplied password.  For each file in each backup fileset, a unique random salt value is generated, and combined with the password via a key derivation function to generate a symmetric encryption key for that file.  The key is then used to encrypt the backup data using AES 256 CBC inline.  For restoration, ndb_restore combines the user supplied password with the per-file salt values to determine the unique decryption key for each file.

Key = KDF(RandomSalt, Password)

EncryptedData = AES_256_CBC(Key, PlainData)

Using an external password in this way allows users to manage backup encryption secrets in alignment with existing policies and systems.


Backup and restore parallelism

Each Ndb backup runs in parallel across all data nodes, with each node producing independent filesets.  At restore time, each ndb_restore instance restores batches of rows concurrently, and multiple ndb_restore instances can be run in parallel, each operating on independent backup filesets.

From the 8.0.16 release onwards, each data node can produce multiple backup filesets - one per Local Data Manager (LDM) component, and ndb_restore restores multiple backup filesets in parallel by default.

In the 8.0.20 release, the ability to use multiple ndb_restore instances to restore slices of a single backup fileset was added.

Triggering a backup

Users trigger a new Ndb backup using the management api (ndb_mgm / mgmapi).  The command in the ndb_mgm shell (version 8.0.22) looks like this :

ndb_mgm > START BACKUP [<backup id>] [ENCRYPT PASSWORD='<password>']
              [SNAPSHOTSTART | SNAPSHOTEND] [NOWAIT | WAIT STARTED | WAIT COMPLETED]

The backup id affects the name of the backup files on disk.  If no backup id is supplied, then the cluster generates one internally.

If an encryption password is supplied, then the generated backup files will be encrypted using this password.  If the RequireEncryptedBackup parameter is set in the cluster configuration, then a password must always be supplied.

SnapshotStart and SnapshotEnd give some control over the choice of the transactionally consistent point in time that the backup represents - either at the start of the backup, before the backup is STARTED, or at the end of the backup, before the backup is COMPLETED.  The default is SnapshotEnd, but SnapshotStart can be useful for coordinating the backup of an Ndb cluster with other clusters and external systems.

The NoWait, Wait Started and Wait Completed options affect when the START BACKUP command itself will return control to the user, with Wait Completed as the default.

Restoring a backup

The ndb_restore tool operates on one or more backup filesets generated by a node at backup time.  The main options are :

• --restore-meta : Read the schema metadata from the backup and create the same schema in the target cluster.
• --restore-data : Read the backup data and log files, and use NdbApi to insert, update and delete data to restore to the backup consistency point.
  

Options can be combined, so that a single ndb_restore invocation performs multiple actions.

All backup filesets redundantly contain the schema metadata, but the --restore-meta step need only be performed once.  This step will recreate the schema as it was at backup time, though in cases where the table distribution was not explicitly set, it will be automatically adjusted to suit the target cluster.

For the --restore-data step, the rows in the backup fileset's DATA file are first inserted into the target cluster, then the entries from the backup fileset's LOG file are idempotently applied to bring the target cluster to a transactionally consistent state.

For a SNAPSHOTEND backup, the backup log contains actions to redo, which are applied in the order they were logged.
For a SNAPSHOTSTART backup, the backup log contains actions to undo, which are applied in the reverse of the order that they were logged.

Where there are schema changes, or transformations applied, these are performed as data is restored.  One exception is where type changes convert to/from/between BLOB types - in this case ndb_restore automatically uses internal staging tables to perform the restore in two steps.

ndb_restore can be used to restore data from backup filesets while a cluster is in operation, with some caveats :

• Data being restored is only consistent when the restore is complete, before which :
  
• Other clients can observe partially restored data
• Constraints may be temporarily violated

Some other relevant ndb_restore options :

• --include-databases
  Database(s) to include
• --exclude-databases
  Database(s) to exclude
• --include-tables
  Table(s) to include
• --exclude-tables
  Table(s) to exclude
• --backup-password
  Supply password used to encrypt the backup
• --disable-indexes
  Temporarily disable all indexes and constraints described in the backup schema metadata
• --rebuild-indexes
  Rebuild all disabled indexes and constraints described in the backup schema metadata
• --restore-epoch
  Generate a pseudo row in the mysql.ndb_apply_status table containing the restore epoch of this backup
• --promote-attributes
  Allow restore to target schema with wider types
• --lossy-conversions
  Allow restore to target schema with narrower types
• --allow-pk-changes
  Allow primary key differences in target schema
• --remap-column
  Apply some data transforms during restore
  

The ndb_restore tool can also be used to show the content of backup filesets in various ways.

Restoration sequences 

ndb_restore is a flexible tool.  Full cluster restoration normally follows a sequence like this.

1. Restore/create schema - serial
   
   ndb_restore --restore-meta ...
   
   or some other means of creating a schema :
   
   • Apply a mysqldump schema dump
   • Execute a SQL script
   • Use ndb_restore --restore-meta, then ALTER the tables
   •  ...
    
   Tables or databases can be included or excluded.
   
   
   
   
2. [ Drop indices and constraints - serial ]
   
   ndb_restore --disable-indexes ...
   
   This step is only required if there are unique or foreign key constraints on the tables being restored.
   
   Note that if the target cluster schema does not reflect the schema in the backup, then extra steps may be needed here to ensure all indices and constraints are disabled and rebuilt.
   
   
   
   
3. Restore data - parallel/serial
   
   ndb_restore --restore-data ...
   
   This step must be performed for each node which generated backup filesets, but these steps can be done serially or in parallel as they are independent to each other.
   
   Tables or databases can be included or excluded.
   
   If the schema in the target cluster is different, then schema flexibility options may be required.
    
   
   
   
4. [ Restore epoch - serial ]
   
   ndb_restore --restore-epoch ...
   
   This step is only necessary when restoring a backup from another cluster which will act as an async replication source.
   
   
    
5. [ Rebuild indices and constraints - serial ]
   
   ndb_restore --rebuild-indexes ...
   
   This step is only necessary when indices and constraints have been disabled (step 2).
   While the step itself must be run serially, internally it makes use of the parallelism of the cluster for fast index and constraint rebuilds.
   
   
   

For restoring a backup from a single data node cluster, with no schema changes, a single command can perform all of these steps in one pass :

ndb_restore --restore-meta --disable-indexes --restore-data --restore-epoch --rebuild-indexes ...

However, it is more common that a backup has come from a multi node cluster, and requires a number of invocations of ndb_restore to fully restore.  This enables efficient parallel restore independent of cluster scale.

FOSDEM 2020

My post-FOSDEM detox has started - despite preparing by reading some survival guides, I hadn't really fathomed the variety and quantity (and quality) of beer that would flow over four days.  On reflection however, the beer flow has been far exceeded by the flow of tech content and conversation.

On Thursday and Friday I attended the pre-FOSDEM MySQL Days fringe event, where there were two tracks of talks and tutorials on MySQL including sessions on :
 - MySQL Server simplification
 - MySQL replication tooling improvements
 - Configuring group replication
 - Troubleshooting group replication
 - Using DNS for loadbalancing and failover
 - Upgrading to MySQL 8.0
 - New hash join implementation
 - Indexing JSON arrays
 - Datetime types
 - Check constraints
 - New VALUES() syntax
 - Security-hardening MySQL
 - Document store
 - MySQL Analytics
 - MySQL Replication performance modelling
 - Machine learning for MySQL service automation
 - Using reduced durability modes
 - Benchmarking
 - Using Vitesse
 - Using InnoDB Cluster

There were sessions from Oracle, Percona, Planetscale, Facebook, TicketSolve, DBdeployer and Uber.

Naturally the highlights for me were the Friday sessions focused on MySQL Ndb Cluster including :

• MySQL Ndb 8.0 101 - Bernd Ocklin
  
  Bernd from Oracle introduced MySQL Ndb Cluster 8.0 which became generally available just last month.  He covered the concepts, history, system capabilities and the recent changes, setting the scene well for the following sessions.
  
  Bernd also discussed MySQL Ndb Cluster in this video interview.

• MySQL NDB 8.0 clusters in your laptop with DBdeployer - Giuseppe Maxia
  
  This was a great session from the Datacharmer Giuseppe, where he talked about DBdeployer, which is a tool for trying out different database software versions, configurations and topologies.  Giuseppe demonstrated how to use DBDeployer to setup two MySQL Ndb Clusters very simply, starting them in under 30 seconds and testing both the synchronous internal replication within each cluster and the asynchronous replication between them.
  
  If you want to experiment with MySQL Cluster then I think DBDeployer looks like the easiest and quickest way to get started.

 

• SQL with MySQL Ndb 8.0 faster than your NoSQL allow - Bernd Ocklin
  
  This session was focused on one of Ndb's traditional strengths - performance.  Bernd presented an array of benchmarks showing that MySQL Ndb Cluster continues to be faster than the NoSQL products invented in response to the claim that 'SQL cannot scale'. 
  
  The presentation had lots of graphs and numbers showing linear scalability, hundreds of millions of reads per second from a single cluster, winning efficiency + performance comparisons on YCSB, scalability of parallel disk I/O and scalability of Ndb as a Hadoop metadata and small file store.
  
  Bernd then gave some insights into the unique architecture of Ndb, which allows it to achieve these numbers.
  
  Finally there were some more graphs and numbers :
  
  First looking at TPC-H query performance as data node internal parallelism scales, and showing how the query processing parallelism in Ndb allows a 2-node distributed (HA) Ndb setup to outperform a single node InnoDB on some queries.
  
  Next looking at DBT2 (TPC-C) scalability as # replicas and # NodeGroups were increased, and also looking at the behaviour of DBT2 when run using Optane memory for system scale.

• Boosting MySQL Ndb Cluster & MySQL InnoDB Cluster with ProxySQL v2 - Marco Tusa
  
  Marco from Percona showed how having a layer of proxies in front of a cluster can simplify providing an HA MySQL service, smooth operations and allow new features not available in MySQL itself.  He laid out architectures for InnoDB Cluster and Ndb Cluster, and noted that Ndb cluster was the only cluster that could scale out writes. 
  
  He enthusiastically followed this up with numbers by showing how adding MySQL Servers to a cluster can allow it to scale beyond the limits imposed by a single MySQLD.

• MySQL Ndb 8.0 Cluster tutorial - Frazer Clement
  
  In my session we stepped through installing, configuring, starting and using a MySQL Cluster, including looking at high availability features, push down joins, foreign keys, JSON, Binlogging, Synchronized privilieges, and online scale-out.
  
  The audience were generous with feedback and questions, complementing my mid-90's style use of a text file, and generally surprised at how easily they had a cluster running on their laptop, even when manually performing all the steps required.

After my tutorial, I was interested in locating and consuming a drink, and luckily the MySQL Community dinner was just starting in the event venue.  This was a very agreeable event, well attended by the wider community including colleagues from 'old MySQL' that I had not seen for over 10 years.  It is great to see that while there are differences of opinion, and competition, we still have more in common than not.

On Saturday the action moved to the FOSDEM conference itself, running in a University campus.  Despite looking at the program I had not quite appreciated the volume of tracks, sessions and experts that would be present and talking in the rooms and the corridors.  The MySQL developer room alone had 17 talks from across the community, one every half hour for eight and a half hours !

I spent most of my time here, but also attended some talks in the other tracks, including some interesting performance talks in the Software Defined Networking room including a nice one on vectorising packet processing in DPDK.  Unfortunately I was unable to attend the run a mainframe on your laptop session in the Retrocomputing room, but luckily everything was being streamed, and the videos are being uploaded here.  While wandering around the University campus I stumbled upon a bar and tried some of the (open-source?) FOSDEM beer.

I will not bore you by enumerating all of the sessions, following beers and conversations, but I can definitely recommend reviewing the schedule and viewing some of the captured videos to get a bit of the experience for yourself.  Even better, if this sounds like your sort of thing, why not see if you can find a way to attend yourself, in 2021?

Many thanks to lefred for organising, running, presenting, photographing and tweeting across all the events.

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?

Transactional memory in 2012

I've been observing the appearance of hardware transactional memory (HTM) systems in the wild with interest. Keen readers might recall posts recalling my work on software for the XA-Core HTM system at Nortel.

The transactional memory concept is unusual in that it seems to have proponents both at the chip level (Azul, Sun's failed Rock SPARC CPU, IBM, Intel) and in the functional language community, most notably Haskell. I suspect the functional language community is motivated by the simplicity of the concurrency abstraction, and the chip community are motivated by the transistor use case. There doesn't seem to be the same demand for it from the vast middle ground of OSs, middleware, applications etc. Does this signify something?

Two things have seemed rather opaque in most coverage of transactional memory systems. The first is when and why it is better than using explicit locking / atomic operations. The second is the time+space properties of actual implementations. Too many expositions are still bound up in the simplicity of the interface to discuss the real benefits and drawbacks. Everybody likes a simplifying abstraction, but not if it is slower or has unpredictable side-effects.

So it was refreshing to read a blog post by Greg Pfister (formerly of IBM), describing an HTM implementation in relative laymans terms. I first read Greg's book 'In Search of Clusters' around the same time I was working on XA-Core (~2000), so Greg has quite some background context. He does not pretend to fully comprehend the implementation, but he asks the right questions. Searching more widely, I came across a discussion of the forthcoming Intel 'Haswell' chip at LWN. The comments here give some insight into the implementation and implications. Hopefully we'll start to hear more about the physical properties of these mechanisms, their sweet spots and limitations.

Eventual Consistency in MySQL Cluster - using epochs

Before getting to the details of how eventual consistency is implemented, we need to look at epochs. Ndb Cluster maintains an internal distributed logical clock known as the epoch, represented as a 64 bit number. This epoch serves a number of internal functions, and is atomically advanced across all data nodes.

Epochs and consistent distributed state

Ndb is a parallel database, with multiple internal transaction coordinator components starting, executing and committing transactions against rows stored in different data nodes. Concurrent transactions only interact where they attempt to lock the same row. This design minimises unnecessary system-wide synchronisation, enabling linear scalability of reads and writes.

The stream of changes made to rows stored at a data node are written to a local Redo log for node and system recovery. The change stream is also published to NdbApi event listeners, including MySQLD servers recording Binlogs. Each node's change stream contains the row changes it was involved in, as committed by multiple transactions, and coordinated by multiple independent transaction coordinators, interleaved in a partial order.

  Incoming independent transactions
 affecting multiple rows

   T3         T4         T7
   T1         T2         T5

    |         |          |
    V         V          V

 --------  --------  --------
 |  1   |  |  2   |  |  3   |
 |  TC  |  |  TC  |  |  TC  |   Data nodes with multiple
 |      |--|      |--|      |   transaction coordinators
 |------|  |------|  |------|   acting on data stored in
 |      |  |      |  |      |       different nodes
 | DATA |  | DATA |  | DATA |
 --------  --------  --------

    |         |          |
    V         V          V

   t4        t4          t3
   t1        t7          t2
   t2        t1          t7
             t5

 Outgoing row change event
  streams by causing
     transaction

These row event streams are generated independently by each data node in a cluster, but to be useful they need to be correlated together. For system recovery from a crash, the data nodes need to recover to a cluster-wide consistent state. A state which contains only whole transactions, and a state which, logically at least, existed at some point in time. This correlation could be done by an analysis of the transaction ids and row dependencies of each recorded row change to determine a valid order for the merged event streams, but this would add significant overhead. Instead, the Cluster uses a distributed logical clock known as the epoch to group large sets of committed transactions together.

Each epoch contains zero or more committed transactions. Each committed transaction is in only one epoch. The epoch clock advances periodically, every 100 milliseconds by default. When it is time for a new epoch to start, a distributed protocol known as the Global Commit Protocol (GCP) results in all of the transaction coordinators in the Cluster agreeing on a point of time in the flow of committing transactions at which to change epoch. This epoch boundary, between the commit of the last transaction in epoch n, and the commit of the first transaction in epoch n+1, is a cluster-wide consistent point in time. Obtaining this consistent point in time requires cluster-wide synchronisation, between all transaction coordinators, but it need only happen periodically.

Furthermore, each node ensures that the all events for epoch n are published before any events for epoch n+1 are published. Effectively the event streams are sorted by epoch number, and the first time a new epoch is encountered signifies a precise epoch boundary.

 Incoming independent transactions

   T3         T4         T7
   T1         T2         T5

    |         |          |
    V         V          V

 --------  --------  --------
 |  1   |  |  2   |  |  3   |
 |  TC  |  |  TC  |  |  TC  |   Data nodes with multiple
 |      |--|      |--|      |   transaction coordinators
 |------|  |------|  |------|   acting on data stored in
 |      |  |      |  |      |      different nodes
 | DATA |  | DATA |  | DATA |
 --------  --------  --------

    |         |          |
    V         V          V

  t4(22)    t4(22)      t3(22)            Epoch 22
  ......    ......      ......
  t1(23)    t7(23)      t2(23)            Epoch 23
  t2(23)    t1(23)      t7(23)
            ......
            t5(24)                        Epoch 24

  Outgoing row change event
  streams by causing transaction
  with epoch numbers in ()

When these independent streams are merge-sorted by epoch number we get a unified change stream. Multiple possible orderings can result.
One Partial ordering is shown here :

      Events      Transactions
               contained in epoch

   t4(22)
   t4(22)      {T4,T3}
   t3(22)

   ......

   t1(23)
   t2(23)
   t7(23)
   t1(23)      {T1, T2, T7}
   t2(23)
   t7(23)

   ......

   t5(24)      {T5}

Note that we can state from this that T4 -> T1 (Happened before), and T1 -> T5. However we cannot say whether T4 -> T3 or T3 -> T4. In epoch 23 we see that the row events resulting from T1, T2 and T7 are interleaved.

Epoch boundaries act as markers in the flow of row events generated by each node, which are then used as consistent points to recover to. Epoch boundaries also allow a single system wide unified transaction log to be generated from each node's row change stream, by merge-sorting the per-node row change streams by epoch number. Note that the order of events within an epoch is still not tightly constrained. As concurrent transactions can only interact via row locks, the order of events on a single row (Table and Primary key value) signifies transaction commit order, but there is by definition no order between transactions affecting independent row sets.

To record a Binlog of Ndb row changes, MySQLD listens to the row change streams arriving from each data node, and merge-sorts them them by epoch into a single, epoch-ordered stream. When all events for a given epoch have been received, MySQLD records a single Binlog transaction containing all row events for that epoch. This Binlog transaction is referred to as an 'Epoch transaction' as it describes all row changes that occurred in an epoch.

Epoch transactions in the Binlog

Epoch transactions in the Binlog have some interesting properties :

• Efficiency : They can be considered a kind of Binlog group commit, where multiple user transactions are recorded in one Binlog (epoch) transaction. As an epoch normally contains 100 milliseconds of row changes from a cluster, this is a significant amortisation.
  
• Consistency : Each epoch transaction contains the row operations which occurred when moving the cluster from epoch boundary consistent state A to epoch boundary consistent state B
  Therefore, when applied as a transaction by a slave, the slave will atomically move from consistent state A to consistent state B
• Inter-epoch ordering : Any row event recorded in epoch n+1 logically happened after every row event in epoch n
• Intra-epoch disorder : Any two row events recorded in epoch n, affecting different rows, may have happened in any order.
• Intra-epoch key-order : Any two row events recorded in epoch n, affecting the same row, happened in the order they are recorded.

The ordering properties show that epochs give only a partial order, enough to subdivide the row change streams into self-consistent chunks. Within an epoch, row changes may be interleaved in any way, except that multiple changes to the same row will be recorded in the order they were committed.

Each epoch transaction contains the row changes for a particular epoch, and that information is recorded in the epoch transaction itself, as an extra WRITE_ROW event on a system table called mysql.ndb_apply_status. This WRITE_ROW event contains the binlogging MySQLD's server id and the epoch number. This event is added so that it will be atomically applied by the Slave along with the rest of the row changes in the epoch transaction, giving an atomically reliable indicator of the replication 'position' of the Slave relative to the Master Cluster in terms of epoch number. As the epoch number is abstracted from the details of a particular Master MySQLD's binlog files and offsets, it can be used to failover to an alternative Master.

We can visualise a MySQL Cluster Binlog as looking something like this. Each Binlog transaction contains one 'artificially generated' WRITE_ROW event at the start, and then RBR row events for all row changes that occurred in that epoch.

    BEGIN
  WRITE_ROW mysql.ndb_apply_status server_id=4, epoch=6998
  WRITE_ROW ...
  UPDATE_ROW ...
  DELETE_ROW ...
  ...
  COMMIT # Consistent state of the database

  BEGIN
  WRITE_ROW mysql.ndb_apply_status server_id=4, epoch=6999
  ...
  COMMIT # Consistent state of the database

  BEGIN
  WRITE_ROW mysql.ndb_apply_status server_id=4, epoch=7000
  ...
  COMMIT # Consistent state of the database
  ...

A series of epoch transactions, each with a special WRITE_ROW event for recording the epoch on the Slave. You can see this structure using the mysqlbinlog tool with the --verbose option.

Rows tagged with last-commit epoch

Each row in a MySQL Cluster stores a hidden metadata column which contains the epoch at which a write to the row was last committed. This information is used internally by the Cluster during node recovery and other operations. The ndb_select_all tool can be used to see the epoch numbers for rows in a table by supplying the --gci or --gci64 options. Note that the per-row epoch is not a row version, as two updates to a row in reasonably quick succession will have the same commit epoch.

Epochs and eventual consistency

Reviewing epochs from the point of view of my previous posts on eventual consistency we see that :

• Epochs provide an incrementing logical clock
• Epochs are recorded in the Binlog, and therefore shipped to Slaves
• Epoch boundaries imply happened-before relationships between events before and after them in the Binlog

The properties mean that epochs are almost perfect for monitoring conflict windows in an active-active circular replication setup, with only a few enhancements required.

I'll describe these enhancements in the next post.

Edit 23/12/11 : Added index

Some MySQL projects I think are cool - Shard-Query

I've already described Justin Swanhart's Flexviews project as something I think is cool. Since then Justin appears to have been working more on Shard-Query which I also think is cool, perhaps even more so than Flexviews.

On the page linked above, Shard-Query is described using the following statements :

"Shard-Query is a distributed parallel query engine for MySQL"

"ShardQuery is a PHP class which is intended to make working with a partitioned dataset easier"

"ParallelPipelining - MPP distributed query engines runs fragments of queries in parallel, combining the results at the end. Like map/reduce except it speaks SQL directly."

The things I like from the above description :

• Distributed
• Parallel
• MySQL
  
• Partitioned
• Fragments
• Map/Reduce
• SQL
  

The things that scare me :

• PHP

My fear of PHP is most likely groundless, based on experiences circa 1998. I suspect it runs much of the real money-earning web, and perhaps brings my scribblings to you. However, the applicability of Shard-Query seems so general, that to actually (or apparently) limit it to web-oriented use cases seems a shame. In any case I am not hipster enough to know which language would be better - OCaml? Dart? Only joking. I suppose that if the MySQL Proxy could do something along these lines then the language debate would be moot.

I am likely to fall foul of the lack-of-original-content test if I quote too much from the Shard-Query website, but the How-it-works section seems relevant here.

How it works

• The query is parsed using http://code.google.com/p/php-sql-parser
• A modified version of the query is executed on each shard.
  
• The queries are executed in parallel using http://gearman.org
  
• The results from each shard are combined together
• A version of the original query is then executed over the combined results
  
• All aggregation is done on the slaves (pushed down)
• Queries with inlists can be made into parallel queries.
• A callback can be used for QueryRouting. You provide a partition column, and a callback which returns information pointing to the correct shard. The most convenient way to do this is with Shard-Key-Mapper

Query rewriting rules

The core of Shard-Query are the query rewriting rules, which Justin introduces in a blog post entitled 8 substitution rules for running SQL in parallel. These transforms and substitutions allow Shard-Query to execute a user supplied query across multiple database shards. A single query (SELECT) can be mapped into a query to be applied to some, or all shards, and further queries to be used to merge the results of the per-shard queries into a final result.

Compared to a single system query, the consistency of the view that the sharded query executes against is less well defined, but this may well be acceptable for some applications.

On a single server

A single MySQL instance offers inter-query parallelism, but currently has very limited intra-query parallelism, Shard-Query can circumvent this by splitting a single query into multiple sharded sub-queries which can run in different MySQLD threads (as they are each submitted by different clients) to give intra-query parallelism. To me this seems more of a cool side effect and proof of reasonable implementation efficiency, than a real reason to use Shard-Query. Perhaps someone out there has the perfect use case for this.

Across multiple servers

The big-name MySQL users scale-out with MySQL, storing subsets of data on separate MySQL instances. Shard-Query allows SQL queries spanning all shards to be executed. This is what scaled-out MySQL has been waiting for.

I don't think it would be a good idea to run heavy traffic through Shard-Query to access a set of sharded MySQL instances yet, but Shard-Query gives a great way to perform occasional queries across all shards. This could be great for reporting and perhaps some light mining for patterns, trends etc. The ability to query across live real time data may be a real gain.

Loose coupling and availability

Scaling out via sharding standalone database servers has many difficulties, but the independence of each shard can benefit availability relative to a more tightly coupled distributed system. The loose coupling of the MySQL instances means that it's far less likely that the failure of one shard will drag others down with it, increasing system availability. Shard-Query can give the loosely coupled shards a smoother facade. The limited set of capabilities that Shard-Query gives over the set of shards may well be more than good enough. Note that 'good enough' is a recurring theme in this 'MySQL projects I think are cool' series. Often 'best' results in expensive or unnecessary compromises as a side-effect of trying to please everybody all the time.

Yet another MySQL sharded scaleout design

Looking at Shard-Query, and the recent MySQL-NoSQL Api developments, it seems like a modern MySQL sharded scaleout design might make use of :

• MySQL SQL Apis (PHP, JDBC, ODBC, Ruby, Python, ....)
  
• NoSQL access mechanisms
  (HandlerSocket, Memcached(1,2))
• ShardQuery for SQL reporting / analysis

Per-instance efficiency can be maximised by using the NoSQL access Apis, single-instance SQL is still available if required for the application, and a global SQL view is also available.

This combination of scalability, efficiency and SQL query-ability could be a sweet spot in the increasingly confusing multi-dimensional space of high throughput distributed databases.

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