Apache Igniteを分散RDBとして使う

# はじめに

Apache Igniteを分散RDBとして使ってみます。
分散RDBとしてApache Igniteを見た時の特徴は以下のとおりです。

- ANSI-99準拠
- スケールアウト可能
- 分散SQL JOINのサポート。各ノードのローカルデータに対してJOINが実行される。
- ACID準拠のトランザクション。MVCCを可能としたトランザクション対応。ただし、v2.7（2018年12月リリース）ではベータ版。


# 環境

以下の環境を使用しました。

- CentOS 7.5
- Apache Ignite 2.7

Apache Igniteの環境は以下の手順で構築済みです。

- [インメモリデータグリッド Apache IgniteをJavaのクライアントからアクセスする](https://qiita.com/mkyz08/items/81a526827a573ea17805)


# テーブルを作成する

まず、SQLLineを使用してクラスタに接続し、コマンドラインからSQL文を実行しテーブルを作成します。
JDBCで接続してテーブルを作成することもできますが、今回はコマンドラインからにしました。
まずは、SQLLineを実行して、クラスタに接続します。

```
# /opt/apache-ignite/bin/sqlline.sh --verbose=true -u jdbc:ignite:thin://127.0.0.1/
issuing: !connect jdbc:ignite:thin://127.0.0.1/ '' '' org.apache.ignite.IgniteJdbcThinDriver
Connecting to jdbc:ignite:thin://127.0.0.1/
Connected to: Apache Ignite (version 2.7.0#20181130-sha1:256ae401)
Driver: Apache Ignite Thin JDBC Driver (version 2.7.0#20181130-sha1:256ae401)
Autocommit status: true
Transaction isolation: TRANSACTION_REPEATABLE_READ
sqlline version 1.3.0
0: jdbc:ignite:thin://127.0.0.1/> 
```

次にテーブルを作成します。
"ATOMICITY=TRANSACTIONAL_SNAPSHOT"と指定することで、SQLトランザクションに対応できます。なお、1つのSQLトランザクションで複数のテーブルを対象にする場合は、対象の全てのテーブルをTRANSACTIONAL_SNAPSHOTモードで作成する必要があります。

```
0: jdbc:ignite:thin://127.0.0.1/> CREATE TABLE City (
   id LONG PRIMARY KEY, name VARCHAR)
   WITH "template=replicated, ATOMICITY=TRANSACTIONAL_SNAPSHOT";
No rows affected (0.483 seconds)
0: jdbc:ignite:thin://127.0.0.1/> CREATE TABLE Person (
   id LONG, name VARCHAR, city_id LONG, PRIMARY KEY (id, city_id))
   WITH "backups=1, affinityKey=city_id, ATOMICITY=TRANSACTIONAL_SNAPSHOT";
No rows affected (0.27 seconds)
```

作成されたテーブルを"!tables"コマンドで確認します。

```
0: jdbc:ignite:thin://127.0.0.1/> !tables
+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------+
|           TABLE_CAT            |          TABLE_SCHEM           |           TABLE_NAME           |           TABLE_TYPE           |            REMARKS             |            TYPE_CA |
+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------+
|                                | PUBLIC                         | CITY                           | TABLE                          |                                |                    |
|                                | PUBLIC                         | PERSON                         | TABLE                          |                                |                    |
+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------------------+--------------------+
0: jdbc:ignite:thin://127.0.0.1/> 
```

次にインデックスを作成します。

```
0: jdbc:ignite:thin://127.0.0.1/> CREATE INDEX idx_city_name ON City (name);
No rows affected (0.079 seconds)
0: jdbc:ignite:thin://127.0.0.1/> 
0: jdbc:ignite:thin://127.0.0.1/> CREATE INDEX idx_person_name ON Person (name);
No rows affected (0.036 seconds)
```

データを投入します。

```
0: jdbc:ignite:thin://127.0.0.1/> INSERT INTO City (id, name) VALUES (1, 'Forest Hill');
1 row affected (0.173 seconds)
0: jdbc:ignite:thin://127.0.0.1/> INSERT INTO City (id, name) VALUES (2, 'Denver');
1 row affected (0.039 seconds)
0: jdbc:ignite:thin://127.0.0.1/> INSERT INTO City (id, name) VALUES (3, 'St. Petersburg');
1 row affected (0.042 seconds)
0: jdbc:ignite:thin://127.0.0.1/> 
0: jdbc:ignite:thin://127.0.0.1/> INSERT INTO Person (id, name, city_id) VALUES (1, 'John Doe', 3);
1 row affected (0.13 seconds)
0: jdbc:ignite:thin://127.0.0.1/> INSERT INTO Person (id, name, city_id) VALUES (2, 'Jane Roe', 2);
1 row affected (0.018 seconds)
0: jdbc:ignite:thin://127.0.0.1/> INSERT INTO Person (id, name, city_id) VALUES (3, 'Mary Major', 1);
1 row affected (0.014 seconds)
0: jdbc:ignite:thin://127.0.0.1/> INSERT INTO Person (id, name, city_id) VALUES (4, 'Richard Miles', 2);
1 row affected (0.008 seconds)
```

SELECT文を実行してみると、以下のように結果が取得できます。

```
0: jdbc:ignite:thin://127.0.0.1/> SELECT p.name, c.name
 FROM Person p, City c
 WHERE p.city_id = c.id;
+--------------------------------+--------------------------------+
|              NAME              |              NAME              |
+--------------------------------+--------------------------------+
| Jane Roe                       | Denver                         |
| Richard Miles                  | Denver                         |
| Mary Major                     | Forest Hill                    |
| John Doe                       | St. Petersburg                 |
+--------------------------------+--------------------------------+
4 rows selected (0.081 seconds)
```

# Javaプログラムからクエリを実行する

JavaからIgniteに接続するためには、JDBC Client DriverかJDBC Driverを使用します。
違いはドキュメントに記載されていますが、どう使い分ければよいのかはよく分からなかったので、とりあえず今回はJDBC Driver（JDBC Thin Driver）を使用してみます。

以下、簡単なサンプルプログラムです。
（メイン関数しかない雑なプログラムですが）

```
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

public class SimpleSQLApp
{
	public static void main(String[] args) throws SQLException {

        try (Connection conn = DriverManager.getConnection("jdbc:ignite:thin://192.168.20.71/")) {
        	conn.setAutoCommit(false);

        	System.out.println("select 1 -> ");
        	try (Statement stmt = conn.createStatement()) {
                try (ResultSet rs =
                    stmt.executeQuery("SELECT p.name, c.name FROM Person p INNER JOIN City c on c.id = p.city_id")) {
                    while (rs.next())
                        System.out.println(rs.getString(1) + ", " + rs.getString(2));
                }
            }

            try (PreparedStatement stmt = conn.prepareStatement("INSERT INTO city (id, name) VALUES (?, ?)")) {
                stmt.setLong(1, 99L);
                stmt.setString(2, "Test Location");
                stmt.executeUpdate();
            }

            try (PreparedStatement stmt =
                conn.prepareStatement("INSERT INTO person (id, name, city_id) values (?, ?, ?)")) {
                stmt.setLong(1, 99L);
                stmt.setString(2, "Test Person");
                stmt.setLong(3, 99L);
                stmt.executeUpdate();
            }
            conn.commit();
//          conn.rollback();

            System.out.println("select 2 -> ");
        	try (Statement stmt = conn.createStatement()) {
                try (ResultSet rs =
                    stmt.executeQuery("SELECT p.name, c.name FROM Person p INNER JOIN City c on c.id = p.city_id")) {
                    while (rs.next())
                        System.out.println(rs.getString(1) + ", " + rs.getString(2));
                }
            }

            try (PreparedStatement stmt = conn.prepareStatement("DELETE FROM city WHERE id = ?")) {
                stmt.setLong(1, 99L);
                stmt.executeUpdate();
            }

            try (PreparedStatement stmt =
                conn.prepareStatement("DELETE FROM person WHERE id = ?")) {
                stmt.setLong(1, 99L);
                stmt.executeUpdate();
            }
            conn.commit();
        }
    }
}
```

# 最後に

かなりあっさりと書きましたが、今回は少し使ってみたかっただけなのでこんなところで終了。

トランザクション処理は1台構成の場合は正常に処理できたのですが、2台構成（クラスタ）にすると、INSERT文の結果が返ってこない現象が発生しました。
色々試してみましたが、結局うまくいかなったので、次のバージョンが出たらまた試してみる予定。

[2019-05-20追記]
JDBC Client Driverでも試してみましたが、トランザクションはサポートしていないと怒られました。
サンプルプログラムは以下のようになります。
注意点は
- テーブル名はスキーマ名をつけてあげないといけない（例：public.Person）
- 設定ファイルが必要（今回はdefault-config.xmlを作成）
- 設定ファイルが必要に関連して「ignite-spring」が必要になる
です。

不明点はcache名を指定しないと"Ouch! Argument is invalid: Cache name must not be null or empty."というエラーが発生したこと。
試しに、cache=SQL_PUBLIC_CITYを指定したところうまく動くようになった。SQL_PUBLIC_PERSIONキャッシュにもアクセスしているが、こちらは指定しなかったのにエラーにならなかった。謎

```
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

import org.apache.ignite.Ignition;

public class SimpleSQLAppJDBCClient
{
	public static void main(String[] args) throws SQLException, ClassNotFoundException {

		Ignition.setClientMode(true);
		Class.forName("org.apache.ignite.IgniteJdbcDriver");

        try (Connection conn =         try (Connection conn = DriverManager.getConnection("jdbc:ignite:cfg://cache=SQL_PUBLIC_CITY@file:///C:/pleiades/workspace/igniteclient-sql-indexing/src/main/java/default-config.xml")) {

        	System.out.println("select 1 -> ");
        	try (Statement stmt = conn.createStatement()) {
                try (ResultSet rs =
                    stmt.executeQuery("SELECT p.name, c.name FROM public.Person p INNER JOIN public.City c on c.id = p.city_id")) {
                    while (rs.next())
                        System.out.println(rs.getString(1) + ", " + rs.getString(2));
                }
            }

            try (PreparedStatement stmt = conn.prepareStatement("INSERT INTO city (id, name) VALUES (?, ?)")) {
                stmt.setLong(1, 99L);
                stmt.setString(2, "Test Location");
                stmt.executeUpdate();
            }

            try (PreparedStatement stmt =
                conn.prepareStatement("INSERT INTO person (id, name, city_id) values (?, ?, ?)")) {
                stmt.setLong(1, 99L);
                stmt.setString(2, "Test Person");
                stmt.setLong(3, 99L);
                stmt.executeUpdate();
            }

            System.out.println("select 2 -> ");
        	try (Statement stmt = conn.createStatement()) {
                try (ResultSet rs =
                    stmt.executeQuery("SELECT p.name, c.name FROM Person p INNER JOIN City c on c.id = p.city_id")) {
                    while (rs.next())
                        System.out.println(rs.getString(1) + ", " + rs.getString(2));
                }
            }

            try (PreparedStatement stmt = conn.prepareStatement("DELETE FROM city WHERE id = ?")) {
                stmt.setLong(1, 99L);
                stmt.executeUpdate();
            }

            try (PreparedStatement stmt =
                conn.prepareStatement("DELETE FROM person WHERE id = ?")) {
                stmt.setLong(1, 99L);
                stmt.executeUpdate();
            }
        }
    }
}
```


# 参考

- [Apache Ignite公式 SQL](https://apacheignite-sql.readme.io/docs/overview)

- [SQL Transactions（公式サイト）](https://apacheignite-sql.readme.io/docs/multiversion-concurrency-control)

- [Apache Ignite上で、SQLを動かして遊ぶ](https://kazuhira-r.hatenablog.com/entry/20180530/1527688167)