JDBC batch updates

Add support for JDBC batch updates. It includes an implementation of
Statement.*Batch(...) as well as PreparedStatement.*Batch() methods.

Under the hood SQLConnection uses the pipelining sending requests one by
one asynchronously and awaiting all of them. There are some issues
regarding vinyl storage engine where execution order are not specified
and DDL statements which are not transactional.

Closes: #62

Author: nicktorwald

README.md

@@ -15,6 +15,7 @@ To get the Java connector for Tarantool 1.6.9, visit
 
 ## Table of contents
 * [Getting started](#getting-started)
+* [JDBC](#JDBC)
 * [Cluster support](#cluster-support)
 * [Where to get help](#where-to-get-help)
 
@@ -170,6 +171,67 @@ System.out.println(template.query("select * from hello_world where hello=:id", C
 
 For more implementation details, see [API documentation](http://tarantool.github.io/tarantool-java/apidocs/index.html).
 
+## JDBC
+
+### Batch updates
+
+`Statement` and `PreparedStatement` objects can be used to submit batch
+updates.
+
+For instance, using `Statement` object:
+
+```java
+Statement statement = connection.createStatement();
+statement.addBatch("INSERT INTO student VALUES (30, 'Joe Jones')");
+statement.addBatch("INSERT INTO faculty VALUES (2, 'Faculty of Chemistry')");
+statement.addBatch("INSERT INTO student_faculty VALUES (30, 2)");
+
+int[] updateCounts = stmt.executeBatch();
+```
+
+or using `PreparedStatement`:
+
+```java
+PreparedStatement stmt = con.prepareStatement("INSERT INTO student VALUES (?, ?)");
+stmt.setInt(1, 30);
+stmt.setString(2, "Michael Korj");
+stmt.addBatch();
+stmt.setInt(1, 40);
+stmt.setString(2, "Linda Simpson");
+stmt.addBatch();
+
+int[] updateCounts = stmt.executeBatch();
+```
+
+The connector uses a pipeliing when it performs a batch request. It means
+each query is asynchronously sent one-by-one in order they were specified
+in the batch.
+
+There are a couple of caveats:
+
+- JDBC spec recommends that *auto-commit* mode should be turned off
+to prevent the driver from committing a transaction when a batch request
+is called. The connector is not support transactions and *auto-commit* is
+always enabled, so each statement from the batch is executed in its own
+transaction.
+
+- DDL operations aren't transactional in Tarantool. In this way, a batch
+like this can produce an undefined behaviour (i.e. second statement can fail
+with an error that `student` table does not exist).
+
+```java
+statement.addBatch("CREATE TABLE student (id INT PRIMARY KEY, name VARCHAR(100))");
+statement.addBatch("INSERT INTO student VALUES (1, 'Alex Smith')");
+```
+
+- If `vinyl` storage engine is used an execution order of batch statements is
+not specified. __NOTE:__ This behaviour is incompatible with JDBC spec in the
+sentence "Batch commands are executed serially (at least logically) in the
+order in which they were added to the batch"
+
+- The driver continues processing the remaining commands in a batch once execution
+of a command fails.
+
 ## Cluster support
 
 To be more fault-tolerant the connector provides cluster extensions. In

src/main/java/org/tarantool/jdbc/SQLBatchResultHolder.java

@@ -0,0 +1,26 @@
+package org.tarantool.jdbc;
+
+import java.util.List;
+
+/**
+ * Wrapper for batch SQL query results.
+ */
+public class SQLBatchResultHolder {
+
+    private final List<SQLResultHolder> results;
+    private final Exception error;
+
+    public SQLBatchResultHolder(List<SQLResultHolder> results, Exception error) {
+        this.results = results;
+        this.error = error;
+    }
+
+    public List<SQLResultHolder> getResults() {
+        return results;
+    }
+
+    public Exception getError() {
+        return error;
+    }
+
+}

src/main/java/org/tarantool/jdbc/SQLConnection.java

@@ -32,15 +32,17 @@
 import java.sql.Savepoint;
 import java.sql.Statement;
 import java.sql.Struct;
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeoutException;
+import java.util.function.Function;
 
 /**
  * Tarantool {@link Connection} implementation.
@@ -525,46 +527,59 @@ public int getNetworkTimeout() throws SQLException {
         return (int) client.getOperationTimeout();
     }
 
-    protected SQLResultHolder execute(long timeout, String sql, Object... args) throws SQLException {
+    protected SQLResultHolder execute(long timeout, SQLQueryHolder query) throws SQLException {
         checkNotClosed();
+        return (useNetworkTimeout(timeout))
+            ? executeWithNetworkTimeout(query)
+            : executeWithQueryTimeout(timeout, query);
+    }
+
+    protected SQLBatchResultHolder executeBatch(long timeout, List<SQLQueryHolder> queries) throws SQLException {
+        checkNotClosed();
+        SQLTarantoolClientImpl.SQLRawOps sqlOps = client.sqlRawOps();
+        SQLBatchResultHolder batchResult = useNetworkTimeout(timeout)
+             ? sqlOps.executeBatch(queries)
+             : sqlOps.executeBatch(timeout, queries);
+
+        return batchResult;
+    }
+
+    private boolean useNetworkTimeout(long timeout) throws SQLException {
         int networkTimeout = getNetworkTimeout();
-        return (timeout == 0 || (networkTimeout > 0 && networkTimeout < timeout))
-            ? executeWithNetworkTimeout(sql, args)
-            : executeWithStatementTimeout(timeout, sql, args);
+        return timeout == 0 || (networkTimeout > 0 && networkTimeout < timeout);
     }
 
-    private SQLResultHolder executeWithNetworkTimeout(String sql, Object... args) throws SQLException {
+    private SQLResultHolder executeWithNetworkTimeout(SQLQueryHolder query) throws SQLException {
         try {
-            return client.sqlRawOps().execute(sql, args);
+            return client.sqlRawOps().execute(query);
         } catch (Exception e) {
             handleException(e);
-            throw new SQLException(formatError(sql, args), e);
+            throw new SQLException(formatError(query), e);
         }
     }
 
     /**
      * Executes a query using a custom timeout.
      *
      * @param timeout query timeout
-     * @param sql     query
-     * @param args    query bindings
+     * @param query   query
      *
      * @return SQL result holder
      *
      * @throws StatementTimeoutException if query execution took more than query timeout
      * @throws SQLException              if any other errors occurred
      */
-    private SQLResultHolder executeWithStatementTimeout(long timeout, String sql, Object... args) throws SQLException {
+    private SQLResultHolder executeWithQueryTimeout(long timeout, SQLQueryHolder query) throws SQLException {
         try {
-            return client.sqlRawOps().execute(timeout, sql, args);
+            return client.sqlRawOps().execute(timeout, query);
         } catch (Exception e) {
             // statement timeout should not affect the current connection
             // but can be handled by the caller side
             if (e.getCause() instanceof TimeoutException) {
-                throw new StatementTimeoutException(formatError(sql, args), e.getCause());
+                throw new StatementTimeoutException(formatError(query), e.getCause());
             }
             handleException(e);
-            throw new SQLException(formatError(sql, args), e);
+            throw new SQLException(formatError(query), e);
         }
     }
 
@@ -708,28 +723,74 @@ private void checkHoldabilitySupport(int holdability) throws SQLException {
     /**
      * Provides error message that contains parameters of failed SQL statement.
      *
-     * @param sql    SQL Text.
-     * @param params Parameters of the SQL statement.
+     * @param query SQL query
      *
      * @return Formatted error message.
      */
-    private static String formatError(String sql, Object... params) {
-        return "Failed to execute SQL: " + sql + ", params: " + Arrays.deepToString(params);
+    private static String formatError(SQLQueryHolder query) {
+        return "Failed to execute SQL: " + query.getQuery() + ", params: " + query.getParams();
     }
 
     static class SQLTarantoolClientImpl extends TarantoolClientImpl {
 
+        private Future<?> executeQuery(SQLQueryHolder queryHolder) {
+            return exec(Code.EXECUTE, Key.SQL_TEXT, queryHolder.getQuery(), Key.SQL_BIND, queryHolder.getParams());
+        }
+
+        private Future<?> executeQuery(SQLQueryHolder queryHolder, long timeoutMillis) {
+            return exec(
+                timeoutMillis, Code.EXECUTE, Key.SQL_TEXT, queryHolder.getQuery(), Key.SQL_BIND, queryHolder.getParams()
+            );
+        }
+
         final SQLRawOps sqlRawOps = new SQLRawOps() {
             @Override
-            public SQLResultHolder execute(String sql, Object... binds) {
-                return (SQLResultHolder) syncGet(exec(Code.EXECUTE, Key.SQL_TEXT, sql, Key.SQL_BIND, binds));
+            public SQLResultHolder execute(SQLQueryHolder query) {
+                return (SQLResultHolder) syncGet(executeQuery(query));
             }
 
             @Override
-            public SQLResultHolder execute(long timeoutMillis, String sql, Object... binds) {
-                return (SQLResultHolder) syncGet(
-                    exec(timeoutMillis, Code.EXECUTE, Key.SQL_TEXT, sql, Key.SQL_BIND, binds)
-                );
+            public SQLResultHolder execute(long timeoutMillis, SQLQueryHolder query) {
+                return (SQLResultHolder) syncGet(executeQuery(query, timeoutMillis));
+            }
+
+            @Override
+            public SQLBatchResultHolder executeBatch(List<SQLQueryHolder> queries) {
+                return executeInternal(queries, (query) -> executeQuery(query));
+            }
+
+            @Override
+            public SQLBatchResultHolder executeBatch(long timeoutMillis, List<SQLQueryHolder> queries) {
+                return executeInternal(queries, (query) -> executeQuery(query, timeoutMillis));
+            }
+
+            private SQLBatchResultHolder executeInternal(List<SQLQueryHolder> queries,
+                                                         Function<SQLQueryHolder, Future<?>> fetcher) {
+                List<Future<?>> sqlFutures = new ArrayList<>();
+                // using queries pipelining to emulate a batch request
+                for (SQLQueryHolder query : queries) {
+                    sqlFutures.add(fetcher.apply(query));
+                }
+                // wait for all the results
+                Exception lastError = null;
+                List<SQLResultHolder> items = new ArrayList<>(queries.size());
+                for (Future<?> future : sqlFutures) {
+                    try {
+                        SQLResultHolder result = (SQLResultHolder) syncGet(future);
+                        if (result.isQueryResult()) {
+                            lastError = new SQLException(
+                                "Result set is not allowed in the batch response",
+                                SQLStates.TOO_MANY_RESULTS.getSqlState()
+                            );
+                        }
+                        items.add(result);
+                    } catch (RuntimeException e) {
+                        // empty result set will be treated as a wrong result
+                        items.add(SQLResultHolder.ofEmptyQuery());
+                        lastError = e;
+                    }
+                }
+                return new SQLBatchResultHolder(items, lastError);
             }
         };
 
@@ -758,9 +819,13 @@ protected void completeSql(TarantoolOp<?> future, TarantoolPacket pack) {
 
         interface SQLRawOps {
 
-            SQLResultHolder execute(String sql, Object... binds);
+            SQLResultHolder execute(SQLQueryHolder query);
+
+            SQLResultHolder execute(long timeoutMillis, SQLQueryHolder query);
+
+            SQLBatchResultHolder executeBatch(List<SQLQueryHolder> queries);
 
-            SQLResultHolder execute(long timeoutMillis, String sql, Object... binds);
+            SQLBatchResultHolder executeBatch(long timeoutMillis, List<SQLQueryHolder> queries);
 
         }
 

src/main/java/org/tarantool/jdbc/SQLDatabaseMetadata.java

@@ -946,7 +946,7 @@ public boolean insertsAreDetected(int type) throws SQLException {
 
     @Override
     public boolean supportsBatchUpdates() throws SQLException {
-        return false;
+        return true;
     }
 
     @Override