Merge pull request #388 from yandex/statement-refactor

Statement refactor

Author: alex-krash

README.md

@@ -22,6 +22,48 @@ JDBC Driver Class:
 
 additionally, if you have a few instances, you can use `BalancedClickhouseDataSource`.
 
+
+### Extended API
+In order to provide non-JDBC complaint data manipulation functionality, proprietary API exists.
+Entry point for API is `ClickHouseStatement#write()` method.
+
+#### Importing file into table
+```java
+import ru.yandex.clickhouse.ClickHouseStatement;
+ClickHouseStatement sth = connection.createStatement();
+sth
+    .write() // Write API entrypoint
+    .table("default.my_table") // where to write data
+    .option("format_csv_delimiter", ";") // specific param
+    .data(new File("/path/to/file.csv"), ClickHouseFormat.CSV) // specify input
+    .send();
+```
+#### Configurable send
+```java
+import ru.yandex.clickhouse.ClickHouseStatement;
+ClickHouseStatement sth = connection.createStatement();
+sth
+    .write()
+    .sql("INSERT INTO default.my_table (a,b,c) VALUES")
+    .data(new MyCustomInputStream(), ClickHouseFormat.JSONEachRow)
+    .addDbParam(ClickHouseQueryParam.MAX_PARALLEL_REPLICAS, 2)
+    .send();
+```
+#### Send data in binary formatd with custom user callback
+```java
+import ru.yandex.clickhouse.ClickHouseStatement;
+ClickHouseStatement sth = connection.createStatement();
+sth.write().send("INSERT INTO test.writer", new ClickHouseStreamCallback() {
+    @Override
+    public void writeTo(ClickHouseRowBinaryStream stream) throws IOException {
+        for (int i = 0; i < 10; i++) {
+            stream.writeInt32(i);
+            stream.writeString("Name " + i);
+        }
+    }
+},
+ClickHouseFormat.RowBinary); // RowBinary or Native are supported
+```
 ### Compiling with maven
 The driver is built with maven.
 `mvn package -DskipTests=true`

pom.xml

@@ -4,7 +4,7 @@
 
     <groupId>ru.yandex.clickhouse</groupId>
     <artifactId>clickhouse-jdbc</artifactId>
-    <version>0.1-SNAPSHOT</version>
+    <version>${project.version}</version>
     <packaging>jar</packaging>
 
     <name>clickhouse-jdbc</name>
@@ -39,10 +39,16 @@
             <name>Yuriy Galitskiy</name>
             <email>[email protected]</email>
         </developer>
+        <developer>
+            <id>krash</id>
+            <name>Alexandr Krasheninnikov</name>
+            <email>[email protected]</email>
+        </developer>
     </developers>
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <project.version>0.1.55</project.version>
     </properties>
 
     <dependencies>

src/main/java/ru/yandex/clickhouse/ClickHouseStatement.java

@@ -14,6 +14,7 @@
 
 
 public interface ClickHouseStatement extends Statement {
+
     ClickHouseResponse executeQueryClickhouseResponse(String sql) throws SQLException;
 
     ClickHouseResponse executeQueryClickhouseResponse(String sql, Map<ClickHouseQueryParam, String> additionalDBParams) throws SQLException;
@@ -40,24 +41,68 @@ ResultSet executeQuery(String sql,
                            List<ClickHouseExternalData> externalData,
                            Map<String, String> additionalRequestParams) throws SQLException;
 
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendStream(InputStream content, String table, Map<ClickHouseQueryParam, String> additionalDBParams) throws SQLException;
 
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendStream(InputStream content, String table) throws SQLException;
 
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendRowBinaryStream(String sql, Map<ClickHouseQueryParam, String> additionalDBParams, ClickHouseStreamCallback callback) throws SQLException;
 
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendRowBinaryStream(String sql, ClickHouseStreamCallback callback) throws SQLException;
 
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendNativeStream(String sql, Map<ClickHouseQueryParam, String> additionalDBParams, ClickHouseStreamCallback callback) throws SQLException;
-    
+
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendNativeStream(String sql, ClickHouseStreamCallback callback) throws SQLException;
 
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendCSVStream(InputStream content, String table, Map<ClickHouseQueryParam, String> additionalDBParams) throws SQLException;
 
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendCSVStream(InputStream content, String table) throws SQLException;
 
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendStreamSQL(InputStream content, String sql, Map<ClickHouseQueryParam, String> additionalDBParams) throws SQLException;
 
+    /**
+     * @see #write()
+     */
+    @Deprecated
     void sendStreamSQL(InputStream content, String sql) throws SQLException;
 
+    /**
+     * Returns extended write-API
+     */
+    Writer write();
 }

src/main/java/ru/yandex/clickhouse/ClickHouseStatementImpl.java

@@ -1,55 +1,42 @@
 package ru.yandex.clickhouse;
 
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.HttpURLConnection;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.SQLWarning;
-import java.util.ArrayList;
-import java.util.EnumMap;
-import java.util.List;
-import java.util.Map;
-import java.util.TimeZone;
-import java.util.UUID;
-
+import com.google.common.base.Strings;
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpResponse;
 import org.apache.http.NameValuePair;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.utils.URIBuilder;
 import org.apache.http.entity.ContentType;
-import org.apache.http.entity.InputStreamEntity;
 import org.apache.http.entity.StringEntity;
 import org.apache.http.entity.mime.MultipartEntityBuilder;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.message.BasicNameValuePair;
 import org.apache.http.util.EntityUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Strings;
-
 import ru.yandex.clickhouse.domain.ClickHouseFormat;
 import ru.yandex.clickhouse.except.ClickHouseException;
 import ru.yandex.clickhouse.except.ClickHouseExceptionSpecifier;
-import ru.yandex.clickhouse.response.ClickHouseLZ4Stream;
-import ru.yandex.clickhouse.response.ClickHouseResponse;
-import ru.yandex.clickhouse.response.ClickHouseResultSet;
-import ru.yandex.clickhouse.response.ClickHouseScrollableResultSet;
-import ru.yandex.clickhouse.response.FastByteArrayOutputStream;
+import ru.yandex.clickhouse.response.*;
 import ru.yandex.clickhouse.settings.ClickHouseProperties;
 import ru.yandex.clickhouse.settings.ClickHouseQueryParam;
 import ru.yandex.clickhouse.util.ClickHouseRowBinaryInputStream;
 import ru.yandex.clickhouse.util.ClickHouseStreamCallback;
-import ru.yandex.clickhouse.util.ClickHouseStreamHttpEntity;
 import ru.yandex.clickhouse.util.Utils;
 import ru.yandex.clickhouse.util.guava.StreamUtils;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.util.*;
+
 
 public class ClickHouseStatementImpl implements ClickHouseStatement {
 
@@ -616,9 +603,7 @@ private InputStream getInputStream(
             requestEntity = entityBuilder.build();
         }
 
-        if (properties.isDecompress()) {
-            requestEntity = new LZ4EntityWrapper(requestEntity, properties.getMaxCompressBufferSize());
-        }
+        requestEntity = applyRequestBodyCompression(requestEntity);
 
         HttpEntity entity = null;
         try {
@@ -775,9 +760,8 @@ public void sendRowBinaryStream(String sql, ClickHouseStreamCallback callback) t
 
     @Override
     public void sendRowBinaryStream(String sql, Map<ClickHouseQueryParam, String> additionalDBParams, ClickHouseStreamCallback callback) throws SQLException {
-        sendStream(
-                new ClickHouseStreamHttpEntity(callback, getConnection().getTimeZone(), properties), sql, ClickHouseFormat.RowBinary, additionalDBParams
-        );
+        write().withDbParams(additionalDBParams)
+                .send(sql, callback, ClickHouseFormat.RowBinary);
     }
 
     @Override
@@ -787,73 +771,81 @@ public void sendNativeStream(String sql, ClickHouseStreamCallback callback) thro
 
     @Override
     public void sendNativeStream(String sql, Map<ClickHouseQueryParam, String> additionalDBParams, ClickHouseStreamCallback callback) throws SQLException {
-        sendStream(
-                new ClickHouseStreamHttpEntity(callback, getConnection().getTimeZone(), properties), sql, ClickHouseFormat.Native, additionalDBParams
-        );
+        write().withDbParams(additionalDBParams)
+                .send(sql, callback, ClickHouseFormat.Native);
     }
 
     @Override
-    public void sendCSVStream(InputStream content, String table, Map<ClickHouseQueryParam, String> additionalDBParams) throws ClickHouseException {
-        String query = "INSERT INTO " + table;
-        sendStream(new InputStreamEntity(content, -1), query, ClickHouseFormat.CSV, additionalDBParams);
+    public void sendCSVStream(InputStream content, String table, Map<ClickHouseQueryParam, String> additionalDBParams) throws SQLException {
+        write()
+                .table(table)
+                .withDbParams(additionalDBParams)
+                .data(content)
+                .format(ClickHouseFormat.CSV)
+                .send();
     }
 
     @Override
-    public void sendCSVStream(InputStream content, String table) throws ClickHouseException {
+    public void sendCSVStream(InputStream content, String table) throws SQLException {
         sendCSVStream(content, table, null);
     }
 
     @Override
-    public void sendStream(InputStream content, String table) throws ClickHouseException {
+    public void sendStream(InputStream content, String table) throws SQLException {
         sendStream(content, table, null);
     }
 
     @Override
     public void sendStream(InputStream content, String table,
-                           Map<ClickHouseQueryParam, String> additionalDBParams) throws ClickHouseException {
-        String query = "INSERT INTO " + table;
-        sendStream(new InputStreamEntity(content, -1), query, ClickHouseFormat.TabSeparated, additionalDBParams);
+                           Map<ClickHouseQueryParam, String> additionalDBParams) throws SQLException {
+        write()
+                .table(table)
+                .data(content)
+                .withDbParams(additionalDBParams)
+                .format(ClickHouseFormat.TabSeparated)
+                .send();
     }
 
+    @Deprecated
     public void sendStream(HttpEntity content, String sql) throws ClickHouseException {
         sendStream(content, sql, ClickHouseFormat.TabSeparated, null);
     }
 
+    @Deprecated
     public void sendStream(HttpEntity content, String sql,
                            Map<ClickHouseQueryParam, String> additionalDBParams) throws ClickHouseException {
         sendStream(content, sql, ClickHouseFormat.TabSeparated, additionalDBParams);
     }
 
     private void sendStream(HttpEntity content, String sql, ClickHouseFormat format,
                             Map<ClickHouseQueryParam, String> additionalDBParams) throws ClickHouseException {
-        sendStreamSQL(content, sql + " FORMAT " + format.name(), additionalDBParams);
+
+        Writer writer = write().format(format).withDbParams(additionalDBParams).sql(sql);
+        sendStream(writer, content);
     }
 
     @Override
     public void sendStreamSQL(InputStream content, String sql,
-                              Map<ClickHouseQueryParam, String> additionalDBParams) throws ClickHouseException {
-        sendStreamSQL(new InputStreamEntity(content, -1), sql, additionalDBParams);
+                              Map<ClickHouseQueryParam, String> additionalDBParams) throws SQLException {
+        write().data(content).sql(sql).withDbParams(additionalDBParams).send();
     }
 
     @Override
-    public void sendStreamSQL(InputStream content, String sql) throws ClickHouseException {
-        sendStreamSQL(new InputStreamEntity(content, -1), sql, null);
+    public void sendStreamSQL(InputStream content, String sql) throws SQLException {
+        write().sql(sql).data(content).send();
     }
 
-    private void sendStreamSQL(HttpEntity content, String sql,
-                            Map<ClickHouseQueryParam, String> additionalDBParams) throws ClickHouseException {
-        // echo -ne '10\n11\n12\n' | POST 'http://localhost:8123/?query=INSERT INTO t FORMAT TabSeparated'
+    void sendStream(Writer writer, HttpEntity content) throws ClickHouseException {
         HttpEntity entity = null;
         try {
-            URI uri = buildRequestUri(null, null, additionalDBParams, null, false);
+
+            URI uri = buildRequestUri(writer.getSql(), null, writer.getAdditionalDBParams(), writer.getRequestParams(), false);
             uri = followRedirects(uri);
-            HttpEntity requestEntity = new BodyEntityWrapper(sql, content);
+
+            content = applyRequestBodyCompression(content);
 
             HttpPost httpPost = new HttpPost(uri);
-            if (properties.isDecompress()) {
-                requestEntity = new LZ4EntityWrapper(requestEntity, properties.getMaxCompressBufferSize());
-            }
-            httpPost.setEntity(requestEntity);
+            httpPost.setEntity(content);
             HttpResponse response = client.execute(httpPost);
             entity = response.getEntity();
             checkForErrorAndThrow(entity, response);
@@ -892,6 +884,13 @@ public boolean isCloseOnCompletion() throws SQLException {
         return closeOnCompletion;
     }
 
+    private HttpEntity applyRequestBodyCompression(final HttpEntity entity) {
+        if (properties.isDecompress()) {
+            return new LZ4EntityWrapper(entity, properties.getMaxCompressBufferSize());
+        }
+        return entity;
+    }
+
     private ClickHouseResultSet createResultSet(InputStream is, int bufferSize, String db, String table, boolean usesWithTotals,
     		ClickHouseStatement statement, TimeZone timezone, ClickHouseProperties properties) throws IOException {
     	if(isResultSetScrollable) {
@@ -916,4 +915,9 @@ private Map<ClickHouseQueryParam, String> addQueryIdTo(Map<ClickHouseQueryParam,
 
         return parameters;
     }
+
+    @Override
+    public Writer write() {
+        return new Writer(this);
+    }
 }