Spider and vertical partition engines with new goodies

The Spider storage engine should be already known to the community. Its version 2.5 has recently been released, with new features, the most important of which is that you can execute remote SQL statements in the backend servers. The method is quite simple. Together with Spider, you also get an UDF that executes SQL code in a remote server. You send a query with parameters saying how to connect to the server, and check the result (1 for success, 0 for failure). If the SQL involves a SELECT, the result can be sent to a temporary table. Simple and effective.

In addition to the Spider engine, Kentoku SHIBA has also created the vertical partitioning engine. Instead of splitting tables by record, you split them by columns. You can define a table with column A and column B, with primary key K, and another table with column C and column D, with primary key K. The vertical partition engine allows you to define a table with columns K, A, B, C, D, which looks to the user like a regular column. The backend tables can be of any engine.
There is a MySQL University session about the Spider and VP engines on November 26th at 15:00 CEST. Free attendance!
The slides are online: Sharding for the masses

MySQL Sandbox and Spider at FrOSCon and OpenSQLCamp

FrOSCon and the OpenSQLCamp are about to start. I am packing for Sankt Augustin, where I will attend the fourth edition of FrOSCon and the second OpenSQLCamp. I will have two sessions, Sharding for the masses, about the Spider storage engine and MySQL Sandbox 3, about one of my favorite tools.

The program is very rich. There will be several tracks in the main event and in the associated conferences. If you have any involvement or simply some curiority in open source matters, You will find something interesting at FrOSCon.

Sharding for the masses: the spider storage engine

In my previous article about the Spider storage engine, I made some tests and I saw that the engine has potential. I did also identify some problems, which were promptly addressed by the author. I have looked at the latest version (0.12), and the results are very much encouraging.

Installing the Spider storage engine is not trivial but not extremely difficult either. My previous article has a step-by-step explanation. Before installing, though, you need to apply one patch to the server, to enable condition pushdown to the partitions engine. What is it about? In short, when you issue a query from the Spider table, with a WHERE clause, the query is passed to a backend database without the WHERE clause. True, it's filtered by partition, but each partition may have a large data set, and this means a full table scan on the backend server. In a few cases, this is not a big deal, because the general performance of the Spider engine is better than a non-partitioned table, but it can be improved.
The lack of conditions pushdown is a problem already known in the Federated engine. With the above patch applied, the Spider engine pushes the condition down to the back end, and the query is executed with the WHERE clause intact.
Let's consider this simple query:

select * from salaries where from_date='1996-01-10' limit 10;

Before applying the patch, the backend database would execute this:

select `emp_no`,`salary`,`from_date`,`to_date`
from `employees`.`salaries` limit 0,9223372036854775807

After the patch, the backend receives and execute the more sensible

select `emp_no`,`salary`,`from_date`,`to_date`
from `employees`.`salaries`
where (`from_date` = '1996-01-10')
limit 0,9223372036854775807

A final observation. While I was testing the latest version, I noticed something that should have been clear from the beginning, i.e. that a spider engine set of backends can have multiple front-ends. Meaning that you can create a Spider table from a different host, and access the same backends that another host is already using. In theory, it means more flexibility when dealing with large amount of data. In practice, we'll have to test it more and see what happens.

Test driving the Spider storage engine - sharding for the masses

At the MySQL Conference 2009 I attended a session about the Spider storage engine, an engine with built-in sharding features. The talk was notable for the speaker wearing a spiderman costume, and for some language barrier that made the talk less enjoyable than it should be. That's a pity, because the engine is very intriguing, and deserves some exploration.

What is the Spider engine, then? In short, it's an extension to the partitioning engine with the ability of connecting to remote servers. Basically, partitions + federated, except that Federated is explicitly removed during the compilation. Additionally, the spider engine should remove current limitations, such as single thread for replication and single source replication, but due to lack of specific documentation, I will limit my current experiment to the sharding features.
The only documentation available is the slide deck from the presentation, and some very reference parameters that come with the source code. I show here what I found by practical inspection.

building the engine

To compile the engine, you need the source code for MySQL 5.1.31 (as required by the documentation, but it works fine with later versions as well).
Download the source code from the launchpad repository and expand it. You will get a ./spider directory, which you need to move under the ./storage directory in the source. Then you compile, with these instructions:

autoconf
automake
./configure \
 --prefix=/usr/local/mysql \
 --localstatedir=/usr/local/mysql/data \
 --libexecdir=/usr/local/mysql/bin \
 --enable-thread-safe-client \
 --enable-local-infile --with-pic \
 --with-fast-mutexes \
 --with-client-ldflags=-static \
 --with-mysqld-ldflags=-static \
 --with-zlib-dir=bundled \
 --with-big-tables --with-ssl \
 --with-readline --with-embedded-server \
 --with-partition --without-innodb \
 --without-ndbcluster \
 --without-archive-storage-engine \
 --without-blackhole-storage-engine \
 --with-csv-storage-engine \
 --without-example-storage-engine \
 --without-federated-storage-engine \
 --with-extra-charsets=complex
make
./scripts/make_binary_distribution

Now we will use the MySQL Sandbox to create a testing environment. Let's start with a simple case: 1 main server and 4 remote ones.

make_sandbox $PWD/mysql-5.1.31-osx10.5-i386.tar.gz

This creates a sandbox under $HOME/sandboxes/msb_5_1_31, which is our main server. Before using it, we need to create some additional tables and load the plugin. (These queries are in the spider instructions, but they are hard to cut and paste. This is much easier for that purpose.)

create table if not exists mysql.spider_xa(
  format_id int not null default 0,
  gtrid_length int not null default 0,
  bqual_length int not null default 0,
  data char(128) not null default '',
  status char(8) not null default '',
  primary key (data, format_id, gtrid_length),
  key idx1 (status)
) engine=MyISAM default charset=utf8 collate=utf8_bin;
create table if not exists mysql.spider_xa_member(
  format_id int not null default 0,
  gtrid_length int not null default 0,
  bqual_length int not null default 0,
  data char(128) not null default '',
  scheme char(64) not null default '',
  host char(64) not null default '',
  port char(5) not null default '',
  socket char(64) not null default '',
  username char(64) not null default '',
  password char(64) not null default '',
  primary key (data, format_id, gtrid_length, host, port, socket)
) engine=MyISAM default charset=utf8 collate=utf8_bin;
create table if not exists mysql.spider_tables(
  db_name char(64) not null default '',
  table_name char(64) not null default '',
  priority bigint not null default 0,
  server char(64) default null,
  scheme char(64) default null,
  host char(64) default null,
  port char(5) default null,
  socket char(64) default null,
  username char(64) default null,
  password char(64) default null,
  tgt_db_name char(64) default null,
  tgt_table_name char(64) default null,
  primary key (db_name, table_name),
  key idx1 (priority)
) engine=MyISAM default charset=utf8 collate=utf8_bin;

install plugin spider soname 'ha_spider.so';
select engine,support,transactions,xa 
from information_schema.engines;
+------------+---------+--------------+------+
| engine     | support | transactions | xa   |
+------------+---------+--------------+------+
| SPIDER     | YES     | YES          | YES  | 
| MRG_MYISAM | YES     | NO           | NO   | 
| CSV        | YES     | NO           | NO   | 
| MyISAM     | DEFAULT | NO           | NO   | 
| MEMORY     | YES     | NO           | NO   | 
+------------+---------+--------------+------+

Spider is now enabled

preparing the remote servers

The servers used as remote shards can be conveniently replaced, for this experiment, by sandboxes. Let's create three of them, to illustrate the concept.

make_multiple_sandbox --group_directory=spider_dir \
  --sandbox_base_port=6000 --check_base_port  5.1.31

Now we have three sandboxes under $HOME/sandboxes/spider_dir, with ports ranging from 6001 to 6003.
What we need to do is to create, in each server, a table with the same structure as the one that is being sharded in the main server.

$ cd $HOME/sandboxes/spider_dir
$ cat tablea.sql 
drop schema if exists   myspider;
create schema myspider;
use myspider;

Create table tbl_a( 
        col_a int, 
        col_b int, 
        primary key(col_a) 
);

$ ./use_all "source tablea.sql"

The main server

Finally, we have all the components in place, we can create the table for the main server.

drop schema if exists   myspider;
create schema myspider;
use myspider;

Create table tbl_a( 
        col_a int, 
        col_b int, 
        primary key(col_a) 
) engine = Spider 
Connection ' table "tbl_a", user "msandbox", password "msandbox" ' 
partition by range( col_a ) 
( 
        partition pt1 values less than (1000) 
          comment 'host "127.0.0.1", port "6001"', 
        partition pt2 values less than (2000)
          comment 'host "127.0.0.1", port "6002"', 
        partition pt3 values less than (MAXVALUE)
          comment 'host "127.0.0.1", port "6003"' 
); 

Compared to classic partitions, there is some new ground to cover. The "CONNECTION" clause is used to define the table name in the remote server. The schema name is assumed to be the same as the one in the main server.
For each partition, we can add a "COMMENT" clause, with the connection parameters. Since we are using sandboxes in the same host, we connect to 127.0.0.1, and use the port corresponding to each sandbox.
From now on, we can use this table almost transparently.

Using the spider engine

# on the main server
./use myspider

insert into tbl_a values (1,1);
Query OK, 1 row affected (0.00 sec)

insert into tbl_a values (1001,2);
Query OK, 1 row affected (0.01 sec)

insert into tbl_a values (2001,3);
Query OK, 1 row affected (0.00 sec)

select * from tbl_a;
+-------+-------+
| col_a | col_b |
+-------+-------+
|     1 |     1 | 
|  1001 |     2 | 
|  2001 |     3 | 
+-------+-------+
3 rows in set (0.01 sec)

So far, so good. No surprises, at least in the surface.
Now, where is the data? A close inspection to the files in the data directory shows that the data is not in the main server.
The data is stored in the "remote" servers, as we can check easily:

# in the spider_dir path
./use_all "select * from myspider.tbl_a"

# server: 1: 
col_a col_b
1 1
# server: 2: 
col_a col_b
1001 2
# server: 3: 
col_a col_b
2001 3

Now, let's apply some curiosity. What happens in the remote server when I insert a row in the main server? Probably the general log can give me an answer.

# spider_dir
./use_all 'set global general_log=1'

# main server
insert into tbl_a values (2002,3);
Query OK, 1 row affected (0.00 sec)

# spider_dir
$ tail node3/data/mysql_sandbox6003.log
090429 17:27:28  299 Connect    msandbox@localhost on  
090429 17:27:42  299 Query  set session sql_log_off = 1 

Ah! No curious people allowed.
Well. This can stop a common user, but not a determined one.
MySQL Proxy to the rescue! There is a Lua script that handles logs.

./node2/proxy_start --proxy-lua-script=$PDW/logs.lua --log-level=warning

Let's change the main table definition:

...
        partition pt3 values less than (MAXVALUE)
        comment 'host "127.0.0.1", port "4040"' 

And now we can see what happens.

# main server
insert into tbl_a values (2004,3);
Query OK, 1 row affected (0.00 sec)

#tail /tmp/mysql/log
2009-04-29 18:03:04    305 -- set session sql_log_off = 1 >{0} 
2009-04-29 18:03:04    305 -- set session autocommit = 1 >{0} 
2009-04-29 18:03:04    305 -- start transaction >{0} 
2009-04-29 18:03:04    305 -- set session transaction isolation level repeatable read >{0} 
2009-04-29 18:03:04    305 -- insert into `myspider`.`tbl_a`(`col_a`,`col_b`)values(2004,3) >{0} 
2009-04-29 18:03:04    305 -- commit >{0} 

Hmmm. I don't like the sight of it. autocommit=1 and then start transaction, set session transaction and commit. At the very least, it's wasting three queries. This needs some explanation from the author, I guess. Let's try some data retrieval.

# main server
select * from tbl_a;
+-------+-------+
| col_a | col_b |
+-------+-------+
|     1 |     1 | 
|  1001 |     2 | 
|  2003 |     3 | 
|  2004 |     3 | 
|  1001 |     2 | 
|  2003 |     3 | 
|  2004 |     3 | 
+-------+-------+
7 rows in set (0.01 sec)

$tail /tmp/mysql.log
2009-04-29 18:01:07    303 -- set session sql_log_off = 1 >{0} 
2009-04-29 18:01:07    303 -- set session autocommit = 1 >{0} 
2009-04-29 18:01:07    303 -- start transaction >{0} 
2009-04-29 18:01:07    303 -- set session transaction isolation level repeatable read >{0} 
2009-04-29 18:01:07    303 -- show table status from `myspider` like 'tbl_a' >{0} 
2009-04-29 18:01:07    303 -- select `col_a`,`col_b` from `myspider`.`tbl_a` limit 0,9223372036854775807 >{0} 
2009-04-29 18:01:07    303 -- commit >{0} 

Scarier than the previous one. The LIMIT clause spells trouble.
And this latest experiment made me try something more ambitious.
I installed a group of 20 sandboxes and loaded the employees test database (4.2 million records), spreading two partitioned tables across the backend servers.
Performance is better than using a single table, but slower than a normal partitioned table on a single server. And I think I know why.

# main server
select count(*) from salaries where from_date between '1995-01-01' and '1995-12-31';
+----------+
| count(*) |
+----------+
|   201637 | 
+----------+
1 row in set (0.76 sec)

$ tail /tmp/mysql.log
2009-04-29 18:09:51    307 -- set session sql_log_off = 1 >{0} 
2009-04-29 18:09:51    307 -- set session autocommit = 1 >{0} 
2009-04-29 18:09:51    307 -- start transaction >{0} 
2009-04-29 18:09:51    307 -- set session transaction isolation level repeatable read >{0} 
2009-04-29 18:09:51    307 -- show table status from `employees` like 'salaries' >{0} 
2009-04-29 18:09:51    307 -- select `emp_no`,`from_date` from `employees`.`salaries` order by `emp_no`,`from_date` limit 0,9223372036854775807 >{0} 

This is definitely a waste. It's a problem that is similar to what is happening when using the Federated engine. But here, we get a "ORDER BY" clause that is unnecessary to say the least.

Bugs

During the tests, I spot at least two serious bugs.
When you drop a partition, the data in the remote server is not removed. If you recreate the partition and insert something, you get a "duplicate key" error.

When you drop a database, the table mysql.spider_tables does not get updated, with the result that you can't recreate the table. with the same name, unless you remove the corresponding entries manually.

That was a lot of information for one session. Please try it and comment. Don't expect me to provide answers to the reasons of the spider engine. I hope the author chimes in and clarifies the muddy matter.