# Using the eXtremeSQL Distributed SQL Engine

> The eXtremeSQL Distributed SQL Engine provides limited support for the database sharding architecture for eXtremeDB Cluster installations. Why is it limited? Most full-blown distributed database engines (such as the one found in Oracle for example) normally create an execution plan based on the query tree and data distribution statistics (or other knowledge of how the data is distributed between shards), that contains “map-reduce” style operations. The eXtremeSQL distributed engine merely executes the query on every node over that node's shard, and consolidates the result sets received from multiple nodes when possible (the consolidation of the results is referred to as merge).

> Sometimes merging the result sets is simply impossible, as it is for example for calculating an average. More often, the engine does not have enough information to make sure that the combined result set is correct. The Distributed SQL Engine takes the most optimistic approach -- it always assumes that the application distributed data between shards and created the SQL query to avoid merging problems.

> Yet with the understanding of the engine's limitations, many application's will benefit from using the distributed engine and improve their overall database access performance dramatically.

> The Distributed SQL Engine sends a query to one of the network nodes, or broadcasts queries to all nodes. A node is specified through the query prefix. In order to control the distribution of data, the application must either load data to each shard locally, or specify the node ID (number) in the `insert` statements. For example:

>         10:insert into T values (...)

> Also the application can explicitly use the current node ID in the `select` condition when selecting out records that are inserted on the specific node (`%#` indicates the current node ID, and `%@` the number of shards). For example:

>         insert into hist_cpvehicleid_jj
>             select * from foreign table (path='/home/usr/shea2.csv', skip=1)
>                 as hist_cpvehicleid_jj
>             where mod(hashcode(fstr_vechileid), %#)=%@;

> If none of those methods are used, the Distributed SQL Engine broadcasts the insert on all nodes. The following query types are supported:

>         select * from T;

>         *:select * from T;   -- similar to the above, run the statement on all nodes

>         N:select * from T;   -- execute the statement on the node N (nodes are enumerated from 1)

>         ?:select * from T;   -- execute the statement on any node. SQL picks up the node
>                         in the round-robin fashion, thus implementing a simple
>                         load-balancing scheme

> Note that, normally, the `select`, `update` and `delete` statements are executed on all nodes, while the `insert` statement adds a record to only one of the shards. Once the query has been executed and the result set is created on each node, the Distributed SQL Engine collects the resulting data sets from all nodes. If the query contains an aggregation or sort clause, or `sequence` functions (statistical functions operating on fields of type `sequence`), then the result sets are merged.

> The following diagram illustrates data flow implemented by the Distributed SQL Engine. Note that the rectangles -- `shard1`,`shard2`, `shard3`, `client1` and `client2`, can be placed on the physical different hosts or on the same host, or in any combination. For example `shard1`, `shard2`, `client1` on one physical node and `shard3` and `client2` on another physical node.

> ## Aggregates

> The Distributed SQL Engine currently supports the following aggregates:

> - `COUNT`
> - `MIN`
> - `MAX`
> - `SUM` (aggregate operand is a column) with or without a '`group by`' clause

> The engine does not currently support:

> - Merging aggregates with the `DISTINCT` qualifier;
> - Aggregates over complex expressions (such as `X+Y`), except if the `sort` by or `group by` columns are included into the result set or the `select` statement;

> > For instance (using `Metatable` as an example):

> >         select Metatable.*,FieldNo+FieldSize as ns from Metatable order by ns;

> > or

> >         select T.*,x+y as xy from T order by xy;

> - Aggregates over hash (`seq_hash_aggregate_*`), except for two scenarios:

>   1. when the data that the hash table is built over belongs to a single shard (exchange in the example below):

> > >         select seq_hash_agg_sum(price,exchange) from Quote;

> > 2. if the sequence is converted into a horizontal representation -- `flattened`, the query can run regardless of the data distribution:

> > >         select flattened  seq_hash_agg_sum(price,exchange) from Quote;

> - The `AVG` aggregate. Unless the groups from different nodes do not overlap. For example the following layout is supported:

> >         Node1:
> >         Symbol Price
> >         AAA      10.0
> >         AAA      12.0
> >         AAA       9.0
> >         BBB      11.0
> >         BBB      10.0
> >         BBB      10.0

> >         Node2:
> >         Symbol Price
> >         CCC       8.0
> >         CCC       7.0
> >         CCC      10.0
> >         DDD     15.0
> >         DDD     14.0
> >         DDD     13.0

> >         select avg(Price) from Quote group by Symbol;

> Other examples of valid and invalid statements are:

> >         select * from T;

> > Supported; the results are concatenated.

> >         select * from T order by y;

> > Supported; sort results from all nodes

> >         select * from T order by x+y;

> > Complex expressions are not currently supported

> >         select sum(x) from T;

> > Supported; the aggregated results are merged

> >         select avg(x) from T;

> > Merge of AVG aggregate is not currently supported

> >         select y,sum(x) from T group by y;

> > Supported; groups and aggregates are merged

> >         select sum(x) from T group by y;

> > Supported; note that the 'group by' columns must be included in the "from" list

> >         select sum(x*2) from T;

> > Complex expressions are not currently supported

> >         select ifnull(sum(x), 0) from T;

> > Supported; aggregate results are merged

> >         select seq_sum(x) from T;

> > Supported; results are merged

> >         select seq_hash_agg_sum(x,y) from T;

> > Merge of the hash aggregates are not currently supported

> >         select flattened seq_hash_agg_sum(x,y) from T;

> > Supported; sorted, split in groups and aggregated

> ## Importing data

> The Distributed SQL Engine supports definition for a sharding condition when data is imported from a `CSV` file or through application code. Sharding of `CSV` imported data is specified through the following SQL statement:

> >         select from foreign table (path='csv-file', skip=n) as PatternTable where distribution-condition

> The distribution-condition contains:

> - an `insert-into-select` statement,
> - a `from` clause specifying the CSV filename,
> - an `as` expression specifying the table to insert into,
> - and a `where` clause specifying the destination node.

> As mentioned above, the Distributed SQL Engine also supports special `"%@" and "%#"` pseudo-parameters. The first corresponds to the node number (zero based); the second is used to specify the total number of nodes. For example:

> >         echo "insert into table_name
> >             select * from foreign table (path='table_name_file.csv', skip=1) as table_name
> >             where mod(instrument_sid/$chunk_size,%#)=%@;" > loadrisk.sql
> >         ./xsql.sh loadrisk.sql

> The `xsql.sh` script invokes the Distributed SQL Engine as follows:

> >         xsql @node1 @node2 ... @nodeN $@

> An application can read input data from a stream (a socket in the example below) or any other source, and insert it into the database through a code fragment similar to the following:

> >         table_name tb;
> >         socket_read(s, &tb);
> >         engine.execute("insert into table_name(starttime, endtime,
> >                         book1, book2, instrument_sid)
> >                     values (%l,%l,%s,%s,%l)",
> >                         tb.starttime, tb.endtime,
> >                         tb.book1, tb.book2, tb.instrument_sid);

> The following code fragment would add a sharding condition to the application's code:

> >         char sql[MAX_SQL_STMT_LEN];
> >         table_name tb;
> >         socket_read(s, &tb);
> >         sprintf(sql, "%d:insert into table_name (starttime,endtime,book1,book2,instrument_sid)
> >         values (%%l,%%l,%%s,%%s,%%l)", tb.instrument_sid%n_nodes);
> >         engine.execute(sql, tb.starttime, tb.endtime, tb.book1, tb.book2, tb.instrument_sid);

> ##

> ## Adding Shards at Runtime

> Sometime it may be necessary to add a shard to an existing distributed network. For example, to go from 3 shards to 4. To do so all clients need to re-connect to the 4 shards. This necessitates closing the Distributed SQL Engine that is connected to the 3 shards and reconnect to all 4.

> Moving or rebalancing the data is the application’s responsibility as there is no right way of automatically redistributing data across shards. Essentially it is necessary to pull data to a client and redistribute from that client.

> One approach is to use a file to output data to. For example, output data from an xSQL client to an external file:

>         XSQL>format CSV
>         XSQL>output mytable.csv
>         SELECT * FROM MYTABLE

> Now the file `mytable.csv` is copied to all shards and inserted into the appropriate table(s):

>         XSQL>INSERT INTO MYTABLE from select * from foreign table (path='mytable.csv') where ...

> Another possible approach is to use the following semantics to select data from the table `T` located on node 1 and insert it to the table T on the node 2:

>         XSQL>create table foo(i integer, s varchar);
>         XSQL>1:insert into foo values (1, 'one');
>         XSQL>1:insert into foo values (2, 'two');
>         XSQL>select * from foo;
>         i        s
>         ----------------------------------------------------------------------
>         1        one
>         2        two
>         Selected records:2
>         XSQL>1:select * from foo;
>         i        s
>         ----------------------------------------------------------------------
>         1        one
>         2        two
>         Selected records:2
>         XSQL>2:select * from foo;
>         i        s
>         ----------------------------------------------------------------------
>         Selected records:0

> Now a new object is inserted into table foo on node2, then the contents of table foo on node1 are inserted on node 2 using the “`1>2`” syntax and the results are displayed, first the aggregate from both nodes, then from the individual nodes:

>         XSQL>2:insert into foo values (3, 'three');
>         XSQL>1>2:select * from foo;
>         XSQL>select * from foo;
>         i        s
>         ----------------------------------------------------------------------
>         1        one
>         2        two
>         3        three
>         1        one
>         2        two

>         Selected records:5
>         XSQL>1:select * from foo;
>         i        s
>         ----------------------------------------------------------------------
>         1        one
>         2        two

>         Selected records:2
>         XSQL>2:select * from foo;
>         i        s
>         ----------------------------------------------------------------------
>         3        three
>         1        one
>         2        two

> ## A Note on Conditions for Sharding

> In order for applications to benefit from the horizontal partitioning of data (i.e. sharding) a few conditions should normally be met. First, the data itself ought to be large enough for the search algorithms to benefit from the reduced data size. Search algorithms integrated with the database runtime are very smart in that they employ indexes to minimize I/O. Generally speaking, even for a simple `tree` lookup to benefit from the reduced data sets, the data sets have to be on the order of tens of gigabytes. The performance of a simple `hash` index algorithm is even less dependent on the database size.

> Second, the underlying hardware should provide the means to handle access to shards in parallel. Normally that means that access to each shard is handled by its own CPU core and that the I/O channels for each shard are separated (for example the shards are physically located on different machines). This way the lookup on each shard is truly executed in parallel. If shards are created on the same host, to achieve best results the number of shards should normally be equal to the number of real (not hyper-threaded) CPU cores and the storage media is organized in a `RAID` type of layout.

> Sharding is not without cost. Although it is theoretically possible to utilize sharding from the native eXtremeDB APIs, in practice this is almost never done. The reason is that it is extremely difficult for applications to combine the result sets received from shards correctly and efficiently. The Distributed SQL Engine implements this functionality by creating execution plans, i.e. algorithms that are an integral part of the engine itself. Furthermore, the Distributed SQL Engine is capable of executing a large subset of SQL queries, but it is not the entire set of SQL that can be run against a local database. For example, for the Distributed SQL Engine to execute `JOIN` operations, the data must be organized in a certain way that may require duplicating some data on all shards. Often the memory or media overhead imposed by the Distributed SQL Engine is quite large. Systems that have a lot of resources to spare to improve search performance are normally server-type setups with a large number of CPU cores, tens or even hundreds of gigabytes of memory and distributed physical I/O subsystems.

> The Distributed SQL Engine requires additional system resources such as memory, semaphores, etc., to provide efficient access to a distributed database. Those resources are normally not available in embedded environments. In our experience applications that run in the context of INTEGRITY OS, VxWorks OS on ARM or embedded MIPS CPUs or similar resource constrained setups never benefit from using distributed SQL. If a careful analysis of specific embedded system constraints and application requirements determines that the Distributed SQL Engine is desired, a custom distribution package can be built on request.

> ## Support for Different Host Languages

> In addition to C and C++, the Distributed SQL Engine can be used from Java, C# and Python.

> ### Java

> To create a Distributed connection in Java, simply instantiate a SqlRemoteConnection object by invoking the following constructor overload:

>         /**
>         * Constructor of the distributed database connection.
>         * @param nodes database nodes (each entry should have format "ADDRESS:PORT")
>         */
>         public SqlRemoteConnection(String[] nodes, int maxAttempts)
>         {
>             engine = openDistributed(nodes);
>         }

> For example, the following code snippet opens the Distributed SQL Engine for sharding on two nodes:

>         static String [] nodes = new String[]{"localhost:40000", "localhost:40001"};
>         SqlRemoteConnection con = new SqlRemoteConnection(nodes);

> Note, that runtime initialization is done in the Database class constructor, so a Database object must be created even if not used.

> ### C# (.NET Framework)

> To create a Distributed connection in C#, simply instantiate a SqlRemoteConnection object by invoking the following constructor overload:

>         /**
>         * Constructor of the distributed database connection.
>         * @param nodes database nodes (each entry should have format "ADDRESS:PORT")
>         */
>         public SqlRemoteConnection(String[] nodes)
>         {
>             engine = OpenDistributed(nodes);
>         }

> For example, the following code snippet opens the Distributed SQL Engine for sharding on two nodes:

>         static String [] nodes = new String[]{"localhost:40000", "localhost:40001"};
>         SqlRemoteConnection con = new SqlRemoteConnection(nodes);

> Note, that runtime initialization is done in the Database class constructor, so a Database object must be created even if not used.

> ### Python

> To create a Distributed connection using Python, open the module method connect with a `tuple` as the first argument. For example:

>         con = exdb.connect(('node1:5001', 'node2:5001', 'node3:5001'))

> Also one can create a SQL server using the Python wrapper. For example:

>         conn = exdb.connect("dbname")
>         # pass engine, port and protocol buffer size.
>         # Note that 64K is not enough if sequences are used
>         server = exdb.SqlServer(conn.engine, 50000, 64*1024)
>         server.start()    # Non-blocking call
>         ...               # Do something else or just wait
>         server.stop()     # Stop server
>         conn.close()

> ## Using sharding with xSQL

> To illustrate using the Distributed SQL Engine with xSQL, consider the following database schema (created via SQL):

>         create table Orders (
>             orderId int primary key,
>             product string,
>             customer string,
>             price double,
>             volume doulbe
>         )

> And the following CSV data in file `order.csv`:

>         orderId|product|customer|price|volume
>         1|A|james|10.0|100
>         2|B|bob|50.0|200
>         3|A|paul|11.0|300
>         4|C|paul|100.0|150
>         5|B|bob|52.0|100
>         6|B|bob|49.0|500
>         7|A|james|11.0|100
>         8|C|paul|105.0|300
>         9|A|bob|12.0|400
>         10|C|james|90.0|200

> ### Create SQL servers

> First, to create several databases and run SQL server on them, we use command-line parameters for xSQL as follows to create three instances of an in-memory database with size of `10 Mb` and start the SQL server listening on ports `10001`, `10002` and `10003`:

>         ./xsql -size 10m -p 10001
>         ./xsql -size 10m -p 10002
>         ./xsql -size 10m -p 10003

> Now we can connect to all three servers using the Distributed SQL engine as xSQL accepts the addresses of the servers as command-line options:

>         ./xsql @127.0.0.1:10001 @127.0.0.1:10002 @127.0.0.1:10003

> Alternatively, we could specify the server addresses in a config file (`client.cfg`) like:

>         {
>             remote_client : [ "127.0.0.1:10001", "127.0.0.1:10002","127.0.0.1:10003"]
>         }

> And invoke xSQL as follows:

>         ./xsql -c client.cfg

> ### Using xSQL interactive mode

> After connecting to servers, xSQL goes to interactive mode. First we create a table on all nodes with command:

>         create table Orders (orderId int primary key, product string,
>                     customer string, price double, volume double);

> By default, the Distributed SQL Engine sends queries to all nodes. So the `create table` statement will be executed by all three nodes. Next we can import and distribute the data across the servers using the following SQL statement:

>         insert into Orders select * from 'order.csv' as Orders where mod(orderId, %#)=%@;

> The pseudoparameters '`%#`' and '`%@`' refer to the total number of nodes and zero-based node ID. In our case '`%#`' equals to 3 and '`%@`' is 0 for the first server (running on port `10001`), 1 for the second server and 2 for the third server. For example, on the second server the statement will be equivalent to:

>         insert into Orders select * from 'order.csv' as Orders where mod(orderId, 3)=1;

> and as a result inserts orders with ID 1, 4, 7 and 10. (Note that if you start SQL servers on different hosts, the file `order.csv` must be accessible on all hosts.)

> Now, to check the data, we select records from all nodes :

>         XSQL>select * from Orders order by orderId;
>         orderId product customer        price   volume
>         ------------------------------------------------------------------------------
>         1       A       james   10.000000       100.000000
>         2       B       bob     50.000000       200.000000
>         3       A       paul    11.000000       300.000000
>         4       C       paul    100.000000      150.000000
>         5       B       bob     52.000000       100.000000
>         6       B       bob     49.000000       500.000000
>         7       A       james   11.000000       100.000000
>         8       C       paul    105.000000      300.000000
>         9       A       bob     12.000000       400.000000
>         10      C       james   90.000000       200.000000

>         Selected records: 10

> To select records from the second server only:

>         XSQL>2:select * from Orders;
>         orderId product customer        price   volume
>         ------------------------------------------------------------------------------
>         1       A       james   10.000000       100.000000
>         4       C       paul    100.000000      150.000000
>         7       A       james   11.000000       100.000000
>         10      C       james   90.000000       200.000000

> To use `group by` and `order by` statements:

>         XSQL>select product, sum(price*volume) as s from Orders group by product;
>         product s
>         ------------------------------------------------------------------------------
>         A       10200.000000
>         B       39700.000000
>         C       64500.000000

>         XSQL>select customer, sum(price*volume) as s from Orders group by customer order by s;
>         customer        s
>         ------------------------------------------------------------------------------
>         james   20100.000000
>         bob     44500.000000
>         paul    49800.000000

> ## Using a Table Qualifier

> Currently a distributed client application cannot use a table qualifier as a prefix in the `order by` or the `group by` column lists. For example, given a table Customer and a column `LastName` defined with a schema like the following:

>         class Customer
>         {
>             uint4 customerKey;
>             string FirstName;
>             string LastName;

>             hash<customerKey> by_customerKey[500000];
>         };

>         class Facts
>         {
>             uint4 customerKey;
>             double price;
>             uint4 Quantity;

>             tree<customerKey> by_customerKey;
>         };

> The following `select` syntax is not supported:

>         SELECT customer.FirstName, customer.LastName,  sum(facts.Quantity) as qty,
>             sum(facts.price) as price, as sales
>         FROM customer, facts
>         GROUP BY customer.FirstName, customer.LastName
>         ORDER BY customer.FirstName, customer.LastName;

> Nor is alternative syntax, like the following (using aliases) supported:

>         SELECT supplier.address as s_address, customer.address as c_address
>         FROM supplier, customer
>         ORDER BY s_address, c_address;

> Nor is the following (specifying the column number):

>         SELECT supplier.address, customer.address
>         FROM supplier, customer
>         ORDER BY 1, 2;