diff --git a/pom.xml b/pom.xml
index 7aa339d4b9a99ed330b49d7cfe01d34b582c3442..84fd4140369f781797aba01352236c8e7da37af0 100644
--- a/pom.xml
+++ b/pom.xml
@@ -30,7 +30,7 @@
<dependency>
<groupId>net.jpountz.lz4</groupId>
<artifactId>lz4</artifactId>
- <version>1.1.2</version>
+ <version>1.3.0</version>
</dependency>
<dependency>
@@ -38,6 +38,19 @@
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.7.3</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <version>2.7.3</version>
+ </dependency>
+
</dependencies>
<build>
diff --git a/src/main/java/ru/yandex/metrika/clickhouse/CHConnection.java b/src/main/java/ru/yandex/metrika/clickhouse/CHConnection.java
index 4246514aeebb120596caaa3d4fe49f1834a0d88a..7636b7c87ee3efaf4fe07d7175f4d768ac4a63ce 100644
--- a/src/main/java/ru/yandex/metrika/clickhouse/CHConnection.java
+++ b/src/main/java/ru/yandex/metrika/clickhouse/CHConnection.java
@@ -1,325 +1,12 @@
package ru.yandex.metrika.clickhouse;
-import org.apache.http.impl.client.CloseableHttpClient;
-import ru.yandex.metrika.clickhouse.copypaste.HttpConnectionProperties;
-import ru.yandex.metrika.clickhouse.util.CHHttpClientBuilder;
-import ru.yandex.metrika.clickhouse.util.LogProxy;
-import ru.yandex.metrika.clickhouse.util.Logger;
-
-import java.io.IOException;
-import java.sql.*;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.Executor;
+import java.sql.Connection;
+import java.sql.SQLException;
/**
- * Created by jkee on 14.03.15.
+ * @author serebrserg
+ * @since 22.03.16
*/
-public class CHConnection implements Connection {
- private static final Logger log = Logger.of(CHStatement.class);
-
-
- private final CloseableHttpClient httpclient;
-
- private final HttpConnectionProperties properties = new HttpConnectionProperties();
-
- private CHDataSource dataSource;
-
- private boolean closed = false;
-
- public CHConnection(String url) {
- this.dataSource = new CHDataSource(url);
- CHHttpClientBuilder clientBuilder = new CHHttpClientBuilder(properties);
- log.debug("new connection");
- httpclient = clientBuilder.buildClient();
- }
-
- @Override
- public Statement createStatement() throws SQLException {
- return LogProxy.wrap(Statement.class, new CHStatement(httpclient, dataSource, properties));
- }
-
- public PreparedStatement createPreparedStatement(String sql) throws SQLException {
- return LogProxy.wrap(PreparedStatement.class, new CHPreparedStatement(httpclient, dataSource, properties, sql));
- }
-
- @Override
- public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
- return createStatement(resultSetType, resultSetConcurrency, ResultSet.CLOSE_CURSORS_AT_COMMIT);
- }
-
- @Override
- public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
- if (resultSetType != ResultSet.TYPE_FORWARD_ONLY && resultSetConcurrency != ResultSet.CONCUR_READ_ONLY
- && resultSetHoldability != ResultSet.CLOSE_CURSORS_AT_COMMIT) {
- throw new SQLFeatureNotSupportedException();
- }
- return createStatement();
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql) throws SQLException {
- return createPreparedStatement(sql);
- }
-
- @Override
- public CallableStatement prepareCall(String sql) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public String nativeSQL(String sql) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setAutoCommit(boolean autoCommit) throws SQLException {
-
- }
-
- @Override
- public boolean getAutoCommit() throws SQLException {
- return false;
- }
-
- @Override
- public void commit() throws SQLException {
-
- }
-
- @Override
- public void rollback() throws SQLException {
-
- }
-
- @Override
- public void close() throws SQLException {
- try {
- httpclient.close();
- closed = true;
- } catch (IOException e) {
- throw new CHException("HTTP client close exception", e);
- }
- }
-
- @Override
- public boolean isClosed() throws SQLException {
- return closed;
- }
-
- @Override
- public DatabaseMetaData getMetaData() throws SQLException {
- return LogProxy.wrap(DatabaseMetaData.class, new CHDatabaseMetadata(dataSource.getUrl(), this));
- }
-
- @Override
- public void setReadOnly(boolean readOnly) throws SQLException {
-
- }
-
- @Override
- public boolean isReadOnly() throws SQLException {
- return false;
- }
-
- @Override
- public void setCatalog(String catalog) throws SQLException {
-
- }
-
- @Override
- public String getCatalog() throws SQLException {
- return null;
- }
-
- @Override
- public void setTransactionIsolation(int level) throws SQLException {
-
- }
-
- @Override
- public int getTransactionIsolation() throws SQLException {
- return 0;
- }
-
- @Override
- public SQLWarning getWarnings() throws SQLException {
- return null;
- }
-
- @Override
- public void clearWarnings() throws SQLException {
-
- }
-
-
- @Override
- public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public Map<String, Class<?>> getTypeMap() throws SQLException {
- return null;
- }
-
- @Override
- public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
-
- }
-
- @Override
- public void setHoldability(int holdability) throws SQLException {
-
- }
-
- @Override
- public int getHoldability() throws SQLException {
- return 0;
- }
-
- @Override
- public Savepoint setSavepoint() throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public Savepoint setSavepoint(String name) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void rollback(Savepoint savepoint) throws SQLException {
-
- }
-
- @Override
- public void releaseSavepoint(Savepoint savepoint) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
- return createPreparedStatement(sql);
- }
-
- @Override
- public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public Clob createClob() throws SQLException {
- return null;
- }
-
- @Override
- public Blob createBlob() throws SQLException {
- return null;
- }
-
- @Override
- public NClob createNClob() throws SQLException {
- return null;
- }
-
- @Override
- public SQLXML createSQLXML() throws SQLException {
- return null;
- }
-
- @Override
- public boolean isValid(int timeout) throws SQLException {
- // todo timeout
- Statement statement = createStatement();
- statement.execute("SELECT 1");
- statement.close();
- // no exception - fine
- return true;
- }
-
- @Override
- public void setClientInfo(String name, String value) throws SQLClientInfoException {
-
- }
-
- @Override
- public void setClientInfo(Properties properties) throws SQLClientInfoException {
-
- }
-
- @Override
- public String getClientInfo(String name) throws SQLException {
- return null;
- }
-
- @Override
- public Properties getClientInfo() throws SQLException {
- return null;
- }
-
- @Override
- public Array createArrayOf(String typeName, Object[] elements) throws SQLException {
- return null;
- }
-
- @Override
- public Struct createStruct(String typeName, Object[] attributes) throws SQLException {
- return null;
- }
-
- @Override
- public <T> T unwrap(Class<T> iface) throws SQLException {
- return null;
- }
-
- @Override
- public boolean isWrapperFor(Class<?> iface) throws SQLException {
- return false;
- }
-
- @Override
- public void setSchema(String schema) throws SQLException {
-
- }
-
- @Override
- public String getSchema() throws SQLException {
- return null;
- }
-
- @Override
- public void abort(Executor executor) throws SQLException {
- this.close();
- }
-
- @Override
- public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
-
- }
-
- @Override
- public int getNetworkTimeout() throws SQLException {
- return 0;
- }
+public interface CHConnection extends Connection {
+ CHStatement createCHStatement() throws SQLException;
}
diff --git a/src/main/java/ru/yandex/metrika/clickhouse/CHConnectionImpl.java b/src/main/java/ru/yandex/metrika/clickhouse/CHConnectionImpl.java
new file mode 100644
index 0000000000000000000000000000000000000000..a4d95a0dd01baae1a50e511849d5699e51254fd3
--- /dev/null
+++ b/src/main/java/ru/yandex/metrika/clickhouse/CHConnectionImpl.java
@@ -0,0 +1,335 @@
+package ru.yandex.metrika.clickhouse;
+
+import org.apache.http.impl.client.CloseableHttpClient;
+import ru.yandex.metrika.clickhouse.copypaste.HttpConnectionProperties;
+import ru.yandex.metrika.clickhouse.util.CHHttpClientBuilder;
+import ru.yandex.metrika.clickhouse.util.LogProxy;
+import ru.yandex.metrika.clickhouse.util.Logger;
+
+import java.io.IOException;
+import java.sql.*;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+/**
+ * @author jkee
+ * @since 14.03.15
+ */
+public class CHConnectionImpl implements CHConnection {
+ private static final Logger log = Logger.of(CHStatementImpl.class);
+
+
+ private final CloseableHttpClient httpclient;
+
+ private final HttpConnectionProperties properties = new HttpConnectionProperties();
+
+ private CHDataSource dataSource;
+
+ private boolean closed = false;
+
+ public CHConnectionImpl(String url) {
+ this.dataSource = new CHDataSource(url);
+ CHHttpClientBuilder clientBuilder = new CHHttpClientBuilder(properties);
+ log.debug("new connection");
+ httpclient = clientBuilder.buildClient();
+ }
+
+ @Override
+ public Statement createStatement() throws SQLException {
+ return LogProxy.wrap(CHStatement.class, new CHStatementImpl(httpclient, dataSource, properties));
+ }
+
+ public CHStatement createCHStatement() throws SQLException {
+ return LogProxy.wrap(CHStatement.class, new CHStatementImpl(httpclient, dataSource, properties));
+ }
+
+ public PreparedStatement createPreparedStatement(String sql) throws SQLException {
+ return LogProxy.wrap(PreparedStatement.class, new CHPreparedStatementImpl(httpclient, dataSource, properties, sql));
+ }
+
+ public CHPreparedStatement createCHPreparedStatement(String sql) throws SQLException {
+ return LogProxy.wrap(CHPreparedStatement.class, new CHPreparedStatementImpl(httpclient, dataSource, properties, sql));
+ }
+
+
+ @Override
+ public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
+ return createStatement(resultSetType, resultSetConcurrency, ResultSet.CLOSE_CURSORS_AT_COMMIT);
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
+ if (resultSetType != ResultSet.TYPE_FORWARD_ONLY && resultSetConcurrency != ResultSet.CONCUR_READ_ONLY
+ && resultSetHoldability != ResultSet.CLOSE_CURSORS_AT_COMMIT) {
+ throw new SQLFeatureNotSupportedException();
+ }
+ return createStatement();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql) throws SQLException {
+ return createPreparedStatement(sql);
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public String nativeSQL(String sql) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setAutoCommit(boolean autoCommit) throws SQLException {
+
+ }
+
+ @Override
+ public boolean getAutoCommit() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public void commit() throws SQLException {
+
+ }
+
+ @Override
+ public void rollback() throws SQLException {
+
+ }
+
+ @Override
+ public void close() throws SQLException {
+ try {
+ httpclient.close();
+ closed = true;
+ } catch (IOException e) {
+ throw new CHException("HTTP client close exception", e);
+ }
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return closed;
+ }
+
+ @Override
+ public DatabaseMetaData getMetaData() throws SQLException {
+ return LogProxy.wrap(DatabaseMetaData.class, new CHDatabaseMetadata(dataSource.getUrl(), this));
+ }
+
+ @Override
+ public void setReadOnly(boolean readOnly) throws SQLException {
+
+ }
+
+ @Override
+ public boolean isReadOnly() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public void setCatalog(String catalog) throws SQLException {
+
+ }
+
+ @Override
+ public String getCatalog() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public void setTransactionIsolation(int level) throws SQLException {
+
+ }
+
+ @Override
+ public int getTransactionIsolation() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+
+ }
+
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public Map<String, Class<?>> getTypeMap() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
+
+ }
+
+ @Override
+ public void setHoldability(int holdability) throws SQLException {
+
+ }
+
+ @Override
+ public int getHoldability() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public Savepoint setSavepoint() throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public Savepoint setSavepoint(String name) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void rollback(Savepoint savepoint) throws SQLException {
+
+ }
+
+ @Override
+ public void releaseSavepoint(Savepoint savepoint) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
+ return createPreparedStatement(sql);
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public Clob createClob() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public Blob createBlob() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public NClob createNClob() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public SQLXML createSQLXML() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public boolean isValid(int timeout) throws SQLException {
+ // todo timeout
+ Statement statement = createStatement();
+ statement.execute("SELECT 1");
+ statement.close();
+ // no exception - fine
+ return true;
+ }
+
+ @Override
+ public void setClientInfo(String name, String value) throws SQLClientInfoException {
+
+ }
+
+ @Override
+ public void setClientInfo(Properties properties) throws SQLClientInfoException {
+
+ }
+
+ @Override
+ public String getClientInfo(String name) throws SQLException {
+ return null;
+ }
+
+ @Override
+ public Properties getClientInfo() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public Array createArrayOf(String typeName, Object[] elements) throws SQLException {
+ return null;
+ }
+
+ @Override
+ public Struct createStruct(String typeName, Object[] attributes) throws SQLException {
+ return null;
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException {
+ return null;
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public void setSchema(String schema) throws SQLException {
+
+ }
+
+ @Override
+ public String getSchema() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public void abort(Executor executor) throws SQLException {
+ this.close();
+ }
+
+ @Override
+ public void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
+
+ }
+
+ @Override
+ public int getNetworkTimeout() throws SQLException {
+ return 0;
+ }
+}
diff --git a/src/main/java/ru/yandex/metrika/clickhouse/CHDriver.java b/src/main/java/ru/yandex/metrika/clickhouse/CHDriver.java
index f0e2fc51b47fbe2359a2cbbeabbb7fadc74d8554..dc34991426e8ff2558b317c6d67b3393ebd6db93 100644
--- a/src/main/java/ru/yandex/metrika/clickhouse/CHDriver.java
+++ b/src/main/java/ru/yandex/metrika/clickhouse/CHDriver.java
@@ -33,9 +33,9 @@ public class CHDriver implements Driver {
}
@Override
- public Connection connect(String url, Properties info) throws SQLException {
+ public CHConnection connect(String url, Properties info) throws SQLException {
logger.info("Creating connection");
- return LogProxy.wrap(Connection.class, new CHConnection(url));
+ return LogProxy.wrap(CHConnection.class, new CHConnectionImpl(url));
}
@Override
diff --git a/src/main/java/ru/yandex/metrika/clickhouse/CHPreparedStatement.java b/src/main/java/ru/yandex/metrika/clickhouse/CHPreparedStatement.java
index 06a37d30c3fb6841699e26bace6814f77c53e03d..872cc52d76462518703d692ef219960782b5f3ba 100644
--- a/src/main/java/ru/yandex/metrika/clickhouse/CHPreparedStatement.java
+++ b/src/main/java/ru/yandex/metrika/clickhouse/CHPreparedStatement.java
@@ -1,435 +1,10 @@
package ru.yandex.metrika.clickhouse;
-import org.apache.http.impl.client.CloseableHttpClient;
-import ru.yandex.metrika.clickhouse.copypaste.HttpConnectionProperties;
-import ru.yandex.metrika.clickhouse.util.Logger;
-
-import java.io.InputStream;
-import java.io.Reader;
-import java.math.BigDecimal;
-import java.math.BigInteger;
-import java.net.URL;
-import java.sql.*;
-import java.sql.Date;
-import java.text.NumberFormat;
-import java.text.SimpleDateFormat;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.ZoneId;
-import java.util.*;
+import java.sql.PreparedStatement;
/**
- * Created by zhur on 14/03/16.
+ * @author serebrserg
+ * @since 22.03.16
*/
-public class CHPreparedStatement extends CHStatement implements PreparedStatement {
- private static final Logger log = Logger.of(CHStatement.class);
-
- private final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
- private final SimpleDateFormat dateTimeFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
-
- final String sql;
- final List<String> sqlParts;
- List<String> binds;
-
- public CHPreparedStatement(CloseableHttpClient client, CHDataSource source,
- HttpConnectionProperties properties, String sql) throws SQLException {
- super(client, source, properties);
- this.sql = sql;
- this.sqlParts = parseSql(sql);
- this.binds = new ArrayList<>(this.sqlParts.size() - 1);
- for (int i = 0; i < this.sqlParts.size()-1; i++) {
- this.binds.add(null);
- }
- clearParameters();
- }
-
- protected static List<String> parseSql(String sql) throws SQLException {
- if (sql == null) {
- throw new SQLException("sql statement can't be null");
- }
-
- List<String> parts = new ArrayList<>();
-
- boolean afterBackSlash = false, inQuotes = false, inBackQuotes = false;
- int partStart = 0;
- for (int i = 0; i < sql.length(); i++) {
- char c = sql.charAt(i);
- if (afterBackSlash) {
- afterBackSlash = false;
- } else if (c == '\\') {
- afterBackSlash = true;
- } else if (c == '\'') {
- inQuotes = !inQuotes;
- } else if (c == '`') {
- inBackQuotes = !inBackQuotes;
- } else if (c == '?' && !inQuotes && !inBackQuotes) {
- parts.add(sql.substring(partStart, i));
- partStart = i+1;
- }
- }
- parts.add(sql.substring(partStart, sql.length()));
-
- return parts;
- }
-
- protected String buildSql() throws SQLException {
- if (sqlParts.size() == 1) {
- return sqlParts.get(0);
- }
-
- for(String b : binds) {
- if (b == null) {
- throw new SQLException("Not all parameters binded");
- }
- }
-
- StringBuilder sb = new StringBuilder(sqlParts.get(0));
- for (int i = 1; i < sqlParts.size(); i++) {
- sb.append(binds.get(i-1));
- sb.append(sqlParts.get(i));
- }
- String sql = sb.toString();
-
- return sql;
- }
-
- @Override
- public ResultSet executeQuery() throws SQLException {
- return super.executeQuery(buildSql());
- }
-
- @Override
- public int executeUpdate() throws SQLException {
- return super.executeUpdate(buildSql());
- }
-
- private void setBind(int parameterIndex, String bind) {
- binds.set(parameterIndex-1, bind);
- }
-
- @Override
- public void setNull(int parameterIndex, int sqlType) throws SQLException {
- setBind(parameterIndex, "NULL");
- }
-
- @Override
- public void setBoolean(int parameterIndex, boolean x) throws SQLException {
- setBind(parameterIndex, x ? "1" : "0");
- }
-
- @Override
- public void setByte(int parameterIndex, byte x) throws SQLException {
- setBind(parameterIndex, Byte.toString(x));
- }
-
- @Override
- public void setShort(int parameterIndex, short x) throws SQLException {
- setBind(parameterIndex, Short.toString(x));
- }
-
- @Override
- public void setInt(int parameterIndex, int x) throws SQLException {
- setBind(parameterIndex, Integer.toString(x));
- }
-
- @Override
- public void setLong(int parameterIndex, long x) throws SQLException {
- setBind(parameterIndex, Long.toString(x));
- }
-
- @Override
- public void setFloat(int parameterIndex, float x) throws SQLException {
- setBind(parameterIndex, Float.toString(x));
- }
-
- @Override
- public void setDouble(int parameterIndex, double x) throws SQLException {
- setBind(parameterIndex, Double.toString(x));
- }
-
- @Override
- public void setBigDecimal(int parameterIndex, BigDecimal x) throws SQLException {
- setBind(parameterIndex, x.toPlainString());
- }
-
- @Override
- public void setString(int parameterIndex, String x) throws SQLException {
- setBind(parameterIndex, CHUtil.quote(x));
- }
-
- @Override
- public void setBytes(int parameterIndex, byte[] x) throws SQLException {
- setBind(parameterIndex, new String(x));
- }
-
- @Override
- public void setDate(int parameterIndex, Date x) throws SQLException {
- setBind(parameterIndex, "toDate('" + dateFormat.format(x) + "')");
- }
-
- @Override
- public void setTime(int parameterIndex, Time x) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- // setBind(parameterIndex, "toDateTime('" + dateTimeFormat.format(x) + "')");
- }
-
- @Override
- public void setTimestamp(int parameterIndex, Timestamp x) throws SQLException {
- setBind(parameterIndex, "toDateTime('" + dateTimeFormat.format(x) + "')");
- }
-
- @Override
- public void setAsciiStream(int parameterIndex, InputStream x, int length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- @Deprecated
- public void setUnicodeStream(int parameterIndex, InputStream x, int length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void setBinaryStream(int parameterIndex, InputStream x, int length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void clearParameters() throws SQLException {
- for (int i = 0; i < binds.size()-1; i++) {
- binds.set(i, null);
- }
- }
-
- @Override
- public void setObject(int parameterIndex, Object x, int targetSqlType) throws SQLException {
- setObject(parameterIndex, x);
-
- }
-
- @Override
- public void setObject(int parameterIndex, Object x) throws SQLException {
- if (x == null) {
- setNull(parameterIndex, Types.OTHER);
- } else {
- if (x instanceof Byte) {
- setInt(parameterIndex, ((Byte) x).intValue());
- } else if (x instanceof String) {
- setString(parameterIndex, (String) x);
- } else if (x instanceof BigDecimal) {
- setBigDecimal(parameterIndex, (BigDecimal) x);
- } else if (x instanceof Short) {
- setShort(parameterIndex, ((Short) x).shortValue());
- } else if (x instanceof Integer) {
- setInt(parameterIndex, ((Integer) x).intValue());
- } else if (x instanceof Long) {
- setLong(parameterIndex, ((Long) x).longValue());
- } else if (x instanceof Float) {
- setFloat(parameterIndex, ((Float) x).floatValue());
- } else if (x instanceof Double) {
- setDouble(parameterIndex, ((Double) x).doubleValue());
- } else if (x instanceof byte[]) {
- setBytes(parameterIndex, (byte[]) x);
- } else if (x instanceof Date) {
- setDate(parameterIndex, (Date) x);
- } else if (x instanceof Time) {
- setTime(parameterIndex, (Time) x);
- } else if (x instanceof Timestamp) {
- setTimestamp(parameterIndex, (Timestamp) x);
- } else if (x instanceof LocalDate) {
- setDate(parameterIndex, Date.valueOf((LocalDate) x));
- } else if (x instanceof LocalDateTime) {
- setTimestamp(parameterIndex, Timestamp.valueOf((LocalDateTime) x));
- } else if (x instanceof Boolean) {
- setBoolean(parameterIndex, ((Boolean) x).booleanValue());
- } else if (x instanceof InputStream) {
- setBinaryStream(parameterIndex, (InputStream) x, -1);
- } else if (x instanceof Blob) {
- setBlob(parameterIndex, (Blob) x);
- } else if (x instanceof Clob) {
- setClob(parameterIndex, (Clob) x);
- } else if (x instanceof BigInteger) {
- setString(parameterIndex, x.toString());
- } else {
- throw new SQLDataException("Can't bind object of class "+x.getClass().getCanonicalName());
- }
- }
- }
-
- @Override
- public boolean execute() throws SQLException {
- return super.execute(buildSql());
- }
-
- @Override
- public void addBatch() throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void setCharacterStream(int parameterIndex, Reader reader, int length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void setRef(int parameterIndex, Ref x) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void setBlob(int parameterIndex, Blob x) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void setClob(int parameterIndex, Clob x) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void setArray(int parameterIndex, Array x) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public ResultSetMetaData getMetaData() throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setDate(int parameterIndex, Date x, Calendar cal) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void setTime(int parameterIndex, Time x, Calendar cal) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void setTimestamp(int parameterIndex, Timestamp x, Calendar cal) throws SQLException {
- throw new SQLFeatureNotSupportedException();
-
- }
-
- @Override
- public void setNull(int parameterIndex, int sqlType, String typeName) throws SQLException {
- setNull(parameterIndex, sqlType);
- }
-
- @Override
- public void setURL(int parameterIndex, URL x) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public ParameterMetaData getParameterMetaData() throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setRowId(int parameterIndex, RowId x) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setNString(int parameterIndex, String value) throws SQLException {
- setString(parameterIndex, value);
- }
-
- @Override
- public void setNCharacterStream(int parameterIndex, Reader value, long length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setNClob(int parameterIndex, NClob value) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setClob(int parameterIndex, Reader reader, long length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setBlob(int parameterIndex, InputStream inputStream, long length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setNClob(int parameterIndex, Reader reader, long length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setSQLXML(int parameterIndex, SQLXML xmlObject) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setObject(int parameterIndex, Object x, int targetSqlType, int scaleOrLength) throws SQLException {
- setObject(parameterIndex, x);
- }
-
- @Override
- public void setAsciiStream(int parameterIndex, InputStream x, long length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setBinaryStream(int parameterIndex, InputStream x, long length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setCharacterStream(int parameterIndex, Reader reader, long length) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setAsciiStream(int parameterIndex, InputStream x) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setBinaryStream(int parameterIndex, InputStream x) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setCharacterStream(int parameterIndex, Reader reader) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setNCharacterStream(int parameterIndex, Reader value) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setClob(int parameterIndex, Reader reader) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setBlob(int parameterIndex, InputStream inputStream) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
-
- @Override
- public void setNClob(int parameterIndex, Reader reader) throws SQLException {
- throw new SQLFeatureNotSupportedException();
- }
+public interface CHPreparedStatement extends PreparedStatement {
}
diff --git a/src/main/java/ru/yandex/metrika/clickhouse/CHPreparedStatementImpl.java b/src/main/java/ru/yandex/metrika/clickhouse/CHPreparedStatementImpl.java
new file mode 100644
index 0000000000000000000000000000000000000000..6898bdf13c0862e80a726bde314009758795a65b
--- /dev/null
+++ b/src/main/java/ru/yandex/metrika/clickhouse/CHPreparedStatementImpl.java
@@ -0,0 +1,434 @@
+package ru.yandex.metrika.clickhouse;
+
+import org.apache.http.impl.client.CloseableHttpClient;
+import ru.yandex.metrika.clickhouse.copypaste.HttpConnectionProperties;
+import ru.yandex.metrika.clickhouse.util.Logger;
+
+import java.io.InputStream;
+import java.io.Reader;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.net.URL;
+import java.sql.*;
+import java.text.SimpleDateFormat;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.List;
+
+/**
+ * Created by zhur on 14/03/16.
+ */
+public class CHPreparedStatementImpl extends CHStatementImpl implements CHPreparedStatement {
+ private static final Logger log = Logger.of(CHStatementImpl.class);
+
+ private final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
+ private final SimpleDateFormat dateTimeFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+ final String sql;
+ final List<String> sqlParts;
+ List<String> binds;
+
+ public CHPreparedStatementImpl(CloseableHttpClient client, CHDataSource source,
+ HttpConnectionProperties properties, String sql) throws SQLException {
+ super(client, source, properties);
+ this.sql = sql;
+ this.sqlParts = parseSql(sql);
+ this.binds = new ArrayList<String>(this.sqlParts.size() - 1);
+ for (int i = 0; i < this.sqlParts.size()-1; i++) {
+ this.binds.add(null);
+ }
+ clearParameters();
+ }
+
+ protected static List<String> parseSql(String sql) throws SQLException {
+ if (sql == null) {
+ throw new SQLException("sql statement can't be null");
+ }
+
+ List<String> parts = new ArrayList<String>();
+
+ boolean afterBackSlash = false, inQuotes = false, inBackQuotes = false;
+ int partStart = 0;
+ for (int i = 0; i < sql.length(); i++) {
+ char c = sql.charAt(i);
+ if (afterBackSlash) {
+ afterBackSlash = false;
+ } else if (c == '\\') {
+ afterBackSlash = true;
+ } else if (c == '\'') {
+ inQuotes = !inQuotes;
+ } else if (c == '`') {
+ inBackQuotes = !inBackQuotes;
+ } else if (c == '?' && !inQuotes && !inBackQuotes) {
+ parts.add(sql.substring(partStart, i));
+ partStart = i+1;
+ }
+ }
+ parts.add(sql.substring(partStart, sql.length()));
+
+ return parts;
+ }
+
+ protected String buildSql() throws SQLException {
+ if (sqlParts.size() == 1) {
+ return sqlParts.get(0);
+ }
+
+ for(String b : binds) {
+ if (b == null) {
+ throw new SQLException("Not all parameters binded");
+ }
+ }
+
+ StringBuilder sb = new StringBuilder(sqlParts.get(0));
+ for (int i = 1; i < sqlParts.size(); i++) {
+ sb.append(binds.get(i-1));
+ sb.append(sqlParts.get(i));
+ }
+ String sql = sb.toString();
+
+ return sql;
+ }
+
+ @Override
+ public ResultSet executeQuery() throws SQLException {
+ return super.executeQuery(buildSql());
+ }
+
+ @Override
+ public int executeUpdate() throws SQLException {
+ return super.executeUpdate(buildSql());
+ }
+
+ private void setBind(int parameterIndex, String bind) {
+ binds.set(parameterIndex-1, bind);
+ }
+
+ @Override
+ public void setNull(int parameterIndex, int sqlType) throws SQLException {
+ setBind(parameterIndex, "NULL");
+ }
+
+ @Override
+ public void setBoolean(int parameterIndex, boolean x) throws SQLException {
+ setBind(parameterIndex, x ? "1" : "0");
+ }
+
+ @Override
+ public void setByte(int parameterIndex, byte x) throws SQLException {
+ setBind(parameterIndex, Byte.toString(x));
+ }
+
+ @Override
+ public void setShort(int parameterIndex, short x) throws SQLException {
+ setBind(parameterIndex, Short.toString(x));
+ }
+
+ @Override
+ public void setInt(int parameterIndex, int x) throws SQLException {
+ setBind(parameterIndex, Integer.toString(x));
+ }
+
+ @Override
+ public void setLong(int parameterIndex, long x) throws SQLException {
+ setBind(parameterIndex, Long.toString(x));
+ }
+
+ @Override
+ public void setFloat(int parameterIndex, float x) throws SQLException {
+ setBind(parameterIndex, Float.toString(x));
+ }
+
+ @Override
+ public void setDouble(int parameterIndex, double x) throws SQLException {
+ setBind(parameterIndex, Double.toString(x));
+ }
+
+ @Override
+ public void setBigDecimal(int parameterIndex, BigDecimal x) throws SQLException {
+ setBind(parameterIndex, x.toPlainString());
+ }
+
+ @Override
+ public void setString(int parameterIndex, String x) throws SQLException {
+ setBind(parameterIndex, CHUtil.quote(x));
+ }
+
+ @Override
+ public void setBytes(int parameterIndex, byte[] x) throws SQLException {
+ setBind(parameterIndex, new String(x));
+ }
+
+ @Override
+ public void setDate(int parameterIndex, Date x) throws SQLException {
+ setBind(parameterIndex, "toDate('" + dateFormat.format(x) + "')");
+ }
+
+ @Override
+ public void setTime(int parameterIndex, Time x) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ // setBind(parameterIndex, "toDateTime('" + dateTimeFormat.format(x) + "')");
+ }
+
+ @Override
+ public void setTimestamp(int parameterIndex, Timestamp x) throws SQLException {
+ setBind(parameterIndex, "toDateTime('" + dateTimeFormat.format(x) + "')");
+ }
+
+ @Override
+ public void setAsciiStream(int parameterIndex, InputStream x, int length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ @Deprecated
+ public void setUnicodeStream(int parameterIndex, InputStream x, int length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void setBinaryStream(int parameterIndex, InputStream x, int length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void clearParameters() throws SQLException {
+ for (int i = 0; i < binds.size()-1; i++) {
+ binds.set(i, null);
+ }
+ }
+
+ @Override
+ public void setObject(int parameterIndex, Object x, int targetSqlType) throws SQLException {
+ setObject(parameterIndex, x);
+
+ }
+
+ @Override
+ public void setObject(int parameterIndex, Object x) throws SQLException {
+ if (x == null) {
+ setNull(parameterIndex, Types.OTHER);
+ } else {
+ if (x instanceof Byte) {
+ setInt(parameterIndex, ((Byte) x).intValue());
+ } else if (x instanceof String) {
+ setString(parameterIndex, (String) x);
+ } else if (x instanceof BigDecimal) {
+ setBigDecimal(parameterIndex, (BigDecimal) x);
+ } else if (x instanceof Short) {
+ setShort(parameterIndex, ((Short) x).shortValue());
+ } else if (x instanceof Integer) {
+ setInt(parameterIndex, ((Integer) x).intValue());
+ } else if (x instanceof Long) {
+ setLong(parameterIndex, ((Long) x).longValue());
+ } else if (x instanceof Float) {
+ setFloat(parameterIndex, ((Float) x).floatValue());
+ } else if (x instanceof Double) {
+ setDouble(parameterIndex, ((Double) x).doubleValue());
+ } else if (x instanceof byte[]) {
+ setBytes(parameterIndex, (byte[]) x);
+ } else if (x instanceof Date) {
+ setDate(parameterIndex, (Date) x);
+ } else if (x instanceof Time) {
+ setTime(parameterIndex, (Time) x);
+ } else if (x instanceof Timestamp) {
+ setTimestamp(parameterIndex, (Timestamp) x);
+ } else if (x instanceof LocalDate) {
+ setDate(parameterIndex, Date.valueOf((LocalDate) x));
+ } else if (x instanceof LocalDateTime) {
+ setTimestamp(parameterIndex, Timestamp.valueOf((LocalDateTime) x));
+ } else if (x instanceof Boolean) {
+ setBoolean(parameterIndex, ((Boolean) x).booleanValue());
+ } else if (x instanceof InputStream) {
+ setBinaryStream(parameterIndex, (InputStream) x, -1);
+ } else if (x instanceof Blob) {
+ setBlob(parameterIndex, (Blob) x);
+ } else if (x instanceof Clob) {
+ setClob(parameterIndex, (Clob) x);
+ } else if (x instanceof BigInteger) {
+ setString(parameterIndex, x.toString());
+ } else {
+ throw new SQLDataException("Can't bind object of class "+x.getClass().getCanonicalName());
+ }
+ }
+ }
+
+ @Override
+ public boolean execute() throws SQLException {
+ return super.execute(buildSql());
+ }
+
+ @Override
+ public void addBatch() throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void setCharacterStream(int parameterIndex, Reader reader, int length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void setRef(int parameterIndex, Ref x) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void setBlob(int parameterIndex, Blob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void setClob(int parameterIndex, Clob x) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void setArray(int parameterIndex, Array x) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public ResultSetMetaData getMetaData() throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setDate(int parameterIndex, Date x, Calendar cal) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void setTime(int parameterIndex, Time x, Calendar cal) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void setTimestamp(int parameterIndex, Timestamp x, Calendar cal) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+
+ }
+
+ @Override
+ public void setNull(int parameterIndex, int sqlType, String typeName) throws SQLException {
+ setNull(parameterIndex, sqlType);
+ }
+
+ @Override
+ public void setURL(int parameterIndex, URL x) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public ParameterMetaData getParameterMetaData() throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setRowId(int parameterIndex, RowId x) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setNString(int parameterIndex, String value) throws SQLException {
+ setString(parameterIndex, value);
+ }
+
+ @Override
+ public void setNCharacterStream(int parameterIndex, Reader value, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setNClob(int parameterIndex, NClob value) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setClob(int parameterIndex, Reader reader, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setBlob(int parameterIndex, InputStream inputStream, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setNClob(int parameterIndex, Reader reader, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setSQLXML(int parameterIndex, SQLXML xmlObject) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setObject(int parameterIndex, Object x, int targetSqlType, int scaleOrLength) throws SQLException {
+ setObject(parameterIndex, x);
+ }
+
+ @Override
+ public void setAsciiStream(int parameterIndex, InputStream x, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setBinaryStream(int parameterIndex, InputStream x, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setCharacterStream(int parameterIndex, Reader reader, long length) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setAsciiStream(int parameterIndex, InputStream x) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setBinaryStream(int parameterIndex, InputStream x) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setCharacterStream(int parameterIndex, Reader reader) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setNCharacterStream(int parameterIndex, Reader value) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setClob(int parameterIndex, Reader reader) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setBlob(int parameterIndex, InputStream inputStream) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ @Override
+ public void setNClob(int parameterIndex, Reader reader) throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+}
diff --git a/src/main/java/ru/yandex/metrika/clickhouse/CHStatement.java b/src/main/java/ru/yandex/metrika/clickhouse/CHStatement.java
index 6893d9f17f06a05dab5e7ead891c165c23bc38b0..b79d164437d17bc44238a7791e5736eed6e92ecb 100644
--- a/src/main/java/ru/yandex/metrika/clickhouse/CHStatement.java
+++ b/src/main/java/ru/yandex/metrika/clickhouse/CHStatement.java
@@ -1,438 +1,16 @@
package ru.yandex.metrika.clickhouse;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.entity.StringEntity;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.util.EntityUtils;
-import ru.yandex.metrika.clickhouse.copypaste.*;
-import ru.yandex.metrika.clickhouse.except.ClickhouseExceptionSpecifier;
-import ru.yandex.metrika.clickhouse.util.CopypasteUtils;
-import ru.yandex.metrika.clickhouse.util.Logger;
+import ru.yandex.metrika.clickhouse.copypaste.ClickhouseResponse;
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.HttpURLConnection;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.sql.*;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
+import java.sql.SQLException;
+import java.sql.Statement;
import java.util.Map;
/**
- * Created by jkee on 14.03.15.
+ * @author serebrserg
+ * @since 22.03.16
*/
-public class CHStatement implements Statement {
-
- private static final Logger log = Logger.of(CHStatement.class);
-
- private final CloseableHttpClient client;
-
- private HttpConnectionProperties properties = new HttpConnectionProperties();
-
- private CHDataSource source;
-
- private CHResultSet currentResult;
-
- private int queryTimeout;
-
- private int maxRows;
-
- private boolean closeOnCompletion;
-
- public CHStatement(CloseableHttpClient client, CHDataSource source,
- HttpConnectionProperties properties) {
- this.client = client;
- this.source = source;
- this.properties = properties;
- }
-
- @Override
- public ResultSet executeQuery(String sql) throws SQLException {
- InputStream is = getInputStream(sql, null, false);
- try {
- currentResult = new CHResultSet(properties.isCompress()
- ? new ClickhouseLZ4Stream(is) : is, properties.getBufferSize(),
- extractDBName(sql),
- extractTableName(sql)
- );
- currentResult.setMaxRows(maxRows);
- return currentResult;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public int executeUpdate(String sql) throws SQLException {
- ResultSet rs = null;
- try {
- rs = executeQuery(sql);
- while (rs.next()) {}
- } finally {
- try { rs.close(); } catch (Exception e) {};
- }
- return 1;
- }
-
- @Override
- public boolean execute(String sql) throws SQLException {
- executeQuery(sql);
- return true;
- }
-
-
-
- @Override
- public void close() throws SQLException {
- if (currentResult != null) {
- currentResult.close();
- }
- }
-
- @Override
- public int getMaxFieldSize() throws SQLException {
- return 0;
- }
-
- @Override
- public void setMaxFieldSize(int max) throws SQLException {
-
- }
-
- @Override
- public int getMaxRows() throws SQLException {
- return maxRows;
- }
-
- @Override
- public void setMaxRows(int max) throws SQLException {
- maxRows = max;
- }
-
- @Override
- public void setEscapeProcessing(boolean enable) throws SQLException {
-
- }
-
- @Override
- public int getQueryTimeout() throws SQLException {
- return queryTimeout;
- }
-
- @Override
- public void setQueryTimeout(int seconds) throws SQLException {
- queryTimeout = seconds;
- }
-
- @Override
- public void cancel() throws SQLException {
-
- }
-
- @Override
- public SQLWarning getWarnings() throws SQLException {
- return null;
- }
-
- @Override
- public void clearWarnings() throws SQLException {
-
- }
-
- @Override
- public void setCursorName(String name) throws SQLException {
-
- }
-
- @Override
- public ResultSet getResultSet() throws SQLException {
- return currentResult;
- }
-
- @Override
- public int getUpdateCount() throws SQLException {
- return 0;
- }
-
- @Override
- public boolean getMoreResults() throws SQLException {
- return false;
- }
-
- @Override
- public void setFetchDirection(int direction) throws SQLException {
-
- }
-
- @Override
- public int getFetchDirection() throws SQLException {
- return 0;
- }
-
- @Override
- public void setFetchSize(int rows) throws SQLException {
-
- }
-
- @Override
- public int getFetchSize() throws SQLException {
- return 0;
- }
-
- @Override
- public int getResultSetConcurrency() throws SQLException {
- return 0;
- }
-
- @Override
- public int getResultSetType() throws SQLException {
- return 0;
- }
-
- @Override
- public void addBatch(String sql) throws SQLException {
-
- }
-
- @Override
- public void clearBatch() throws SQLException {
-
- }
-
- @Override
- public int[] executeBatch() throws SQLException {
- return new int[0];
- }
-
- @Override
- public Connection getConnection() throws SQLException {
- return null;
- }
-
- @Override
- public boolean getMoreResults(int current) throws SQLException {
- return false;
- }
-
- @Override
- public ResultSet getGeneratedKeys() throws SQLException {
- return null;
- }
-
- @Override
- public int executeUpdate(String sql, int autoGeneratedKeys) throws SQLException {
- return 0;
- }
-
- @Override
- public int executeUpdate(String sql, int[] columnIndexes) throws SQLException {
- return 0;
- }
-
- @Override
- public int executeUpdate(String sql, String[] columnNames) throws SQLException {
- return 0;
- }
-
- @Override
- public boolean execute(String sql, int autoGeneratedKeys) throws SQLException {
- return false;
- }
-
- @Override
- public boolean execute(String sql, int[] columnIndexes) throws SQLException {
- return false;
- }
-
- @Override
- public boolean execute(String sql, String[] columnNames) throws SQLException {
- return false;
- }
-
- @Override
- public int getResultSetHoldability() throws SQLException {
- return 0;
- }
-
- @Override
- public boolean isClosed() throws SQLException {
- return false;
- }
-
- @Override
- public void setPoolable(boolean poolable) throws SQLException {
-
- }
-
- @Override
- public boolean isPoolable() throws SQLException {
- return false;
- }
-
- @Override
- public <T> T unwrap(Class<T> iface) throws SQLException {
- return null;
- }
-
- @Override
- public boolean isWrapperFor(Class<?> iface) throws SQLException {
- return false;
- }
-
- public static String clickhousifySql(String sql) {
- return clickhousifySql(sql, "TabSeparatedWithNamesAndTypes");
- }
-
- public static String clickhousifySql(String sql, String format) {
- sql = sql.trim();
- if (!sql.replace(";", "").trim().endsWith(" TabSeparatedWithNamesAndTypes")
- && !sql.replace(";", "").trim().endsWith(" TabSeparated")
- && !sql.replace(";", "").trim().endsWith(" JSONCompact")) {
- if (sql.endsWith(";")) sql = sql.substring(0, sql.length() - 1);
- sql += " FORMAT " + format + ';';
- }
- return sql;
- }
-
- private String extractTableName(String sql) {
- String s = extractDBAndTableName(sql);
- if (s.contains(".")) {
- return s.substring(s.indexOf(".") + 1);
- } else return s;
- }
-
- private String extractDBName(String sql) {
- String s = extractDBAndTableName(sql);
- if (s.contains(".")) {
- return s.substring(0, s.indexOf("."));
- } else {
- return source.getDatabase();
- }
- }
-
- private String extractDBAndTableName(String sql) {
- // паршивый код, надо пиÑать или найти нормальный парÑер
- if (CopypasteUtils.startsWithIgnoreCase(sql, "select")) {
- String withoutStrings = CopypasteUtils.retainUnquoted(sql, '\'');
- int fromIndex = withoutStrings.indexOf("from");
- if (fromIndex == -1) fromIndex = withoutStrings.indexOf("FROM");
- if (fromIndex != -1) {
- String fromFrom = withoutStrings.substring(fromIndex);
- String fromTable = fromFrom.substring("from".length()).trim();
- return fromTable.split(" ")[0];
- }
- }
- if (CopypasteUtils.startsWithIgnoreCase(sql, "desc")) {
- return "system.columns"; // bullshit
- }
- if (CopypasteUtils.startsWithIgnoreCase(sql, "show")) {
- return "system.tables"; // bullshit
- }
- return "system.unknown";
- }
-
- private InputStream getInputStream(String sql,
- Map<String, String> additionalClickHouseDBParams,
- boolean ignoreDatabase
- ) throws CHException {
- sql = clickhousifySql(sql);
- log.debug("Executing SQL: " + sql);
- URI uri = null;
- try {
- Map<String, String> params = getParams(false);
- if (additionalClickHouseDBParams != null && !additionalClickHouseDBParams.isEmpty()) {
- params.putAll(additionalClickHouseDBParams);
- }
- List<String> paramPairs = new ArrayList<String>();
- for (Map.Entry<String, String> entry : params.entrySet()) {
- paramPairs.add(entry.getKey() + '=' + entry.getValue());
- }
- String query = CopypasteUtils.join(paramPairs, '&');
- uri = new URI("http", null, source.getHost(), source.getPort(),
- "/", query, null);
- } catch (URISyntaxException e) {
- log.error("Mailformed URL: " + e.getMessage());
- throw new IllegalStateException("illegal configuration of db");
- }
- log.debug("Request url: " + uri);
- HttpPost post = new HttpPost(uri);
- post.setEntity(new StringEntity(sql, CopypasteUtils.UTF_8));
- HttpEntity entity = null;
- InputStream is = null;
- try {
- HttpResponse response = client.execute(post);
- entity = response.getEntity();
- if (response.getStatusLine().getStatusCode() != HttpURLConnection.HTTP_OK) {
- String chMessage = null;
- try {
- InputStream messageStream = entity.getContent();
- if (properties.isCompress()) {
- messageStream = new ClickhouseLZ4Stream(messageStream);
- }
- chMessage = CopypasteUtils.toString(messageStream);
- } catch (IOException e) {
- chMessage = "error while read response "+ e.getMessage();
- }
- EntityUtils.consumeQuietly(entity);
- throw ClickhouseExceptionSpecifier.specify(chMessage, source.getHost(), source.getPort());
- }
- if (entity.isStreaming()) {
- is = entity.getContent();
- } else {
- FastByteArrayOutputStream baos = new FastByteArrayOutputStream();
- entity.writeTo(baos);
- is = baos.convertToInputStream();
- }
- return is;
- } catch (IOException e) {
- log.info("Error during connection to " + source + ", reporting failure to data source, message: " + e.getMessage());
- EntityUtils.consumeQuietly(entity);
- try { if (is != null) is.close(); } catch (IOException ignored) { }
- log.info("Error sql: " + sql);
- throw new CHException("Unknown IO exception", e);
- }
- }
-
- public Map<String, String> getParams(boolean ignoreDatabase) {
- Map<String, String> params = new HashMap<String, String>();
- //в clickhouse бывают таблички без базы (т.е. в базе default)
- if (!CopypasteUtils.isBlank(source.getDatabase()) && !ignoreDatabase) {
- params.put("database", source.getDatabase());
- }
- if (properties.isCompress()) {
- params.put("compress", "1");
- }
- // нам вÑегда нужны min и max в ответе
- params.put("extremes", "1");
- if (CopypasteUtils.isBlank(properties.getProfile())) {
- if (properties.getMaxThreads() != null)
- params.put("max_threads", String.valueOf(properties.getMaxThreads()));
- // да, там в Ñекундах
- params.put("max_execution_time", String.valueOf((properties.getSocketTimeout() + properties.getDataTransferTimeout()) / 1000));
- if (properties.getMaxBlockSize() != null) {
- params.put("max_block_size", String.valueOf(properties.getMaxBlockSize()));
- }
- } else {
- params.put("profile", properties.getProfile());
- }
- //в ÐºÐ»Ð¸ÐºÑ…Ð°ÑƒÑ Ð¸Ð½Ð¾Ð³Ð´Ð° бывает user
- if (properties.getUser() != null) {
- params.put("user", properties.getUser());
- }
- return params;
- }
-
- @Override
- public void closeOnCompletion() throws SQLException {
- closeOnCompletion = true;
- }
-
- @Override
- public boolean isCloseOnCompletion() throws SQLException {
- return closeOnCompletion;
- }
+public interface CHStatement extends Statement {
+ ClickhouseResponse executeQueryClickhouseResponse(String sql) throws SQLException;
+ ClickhouseResponse executeQueryClickhouseResponse(String sql, Map<String, String> additionalDBParams, boolean ignoreDatabase) throws SQLException;
}
diff --git a/src/main/java/ru/yandex/metrika/clickhouse/CHStatementImpl.java b/src/main/java/ru/yandex/metrika/clickhouse/CHStatementImpl.java
new file mode 100644
index 0000000000000000000000000000000000000000..97616da0da0247bb973214895d2334cf5e7a5f80
--- /dev/null
+++ b/src/main/java/ru/yandex/metrika/clickhouse/CHStatementImpl.java
@@ -0,0 +1,474 @@
+package ru.yandex.metrika.clickhouse;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.util.EntityUtils;
+
+
+import ru.yandex.metrika.clickhouse.copypaste.*;
+import ru.yandex.metrika.clickhouse.except.ClickhouseExceptionSpecifier;
+import ru.yandex.metrika.clickhouse.util.CopypasteUtils;
+import ru.yandex.metrika.clickhouse.util.Logger;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.HttpURLConnection;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.sql.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Created by jkee on 14.03.15.
+ */
+public class CHStatementImpl implements CHStatement {
+
+ private static final Logger log = Logger.of(CHStatementImpl.class);
+
+ private final CloseableHttpClient client;
+
+ private HttpConnectionProperties properties = new HttpConnectionProperties();
+
+ private CHDataSource source;
+
+ private CHResultSet currentResult;
+
+ private int queryTimeout;
+
+ private int maxRows;
+
+ private boolean closeOnCompletion;
+
+ private ObjectMapper objectMapper;
+
+ public CHStatementImpl(CloseableHttpClient client, CHDataSource source,
+ HttpConnectionProperties properties) {
+ this.client = client;
+ this.source = source;
+ this.properties = properties;
+
+ objectMapper = new ObjectMapper();
+ objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ }
+
+ @Override
+ public ResultSet executeQuery(String sql) throws SQLException {
+ InputStream is = getInputStream(sql, null, false);
+ try {
+ currentResult = new CHResultSet(properties.isCompress()
+ ? new ClickhouseLZ4Stream(is) : is, properties.getBufferSize(),
+ extractDBName(sql),
+ extractTableName(sql)
+ );
+ currentResult.setMaxRows(maxRows);
+ return currentResult;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public ClickhouseResponse executeQueryClickhouseResponse(String sql) throws SQLException {
+ return executeQueryClickhouseResponse(sql, null, false);
+ }
+
+ public ClickhouseResponse executeQueryClickhouseResponse(String sql, Map<String, String> additionalDBParams, boolean ignoreDatabase) throws SQLException {
+ InputStream is = getInputStream(clickhousifySql(sql, "JSONCompact"), additionalDBParams, ignoreDatabase);
+ try {
+ byte[] bytes = null;
+ try {
+ if (properties.isCompress()){
+ bytes = CopypasteUtils.toByteArray(new ClickhouseLZ4Stream(is));
+ } else {
+ bytes = CopypasteUtils.toByteArray(is);
+ }
+ return objectMapper.readValue(bytes, ClickhouseResponse.class);
+ } catch (IOException e) {
+ if (bytes != null) log.warn("Wrong json: "+new String(bytes));
+ throw e;
+ }
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ if (is != null) try {is.close();} catch (IOException ignored) { }
+ }
+ }
+
+ @Override
+ public int executeUpdate(String sql) throws SQLException {
+ ResultSet rs = null;
+ try {
+ rs = executeQuery(sql);
+ //noinspection StatementWithEmptyBody
+ while (rs.next()) {}
+ } finally {
+ try { if (rs != null) rs.close(); } catch (Exception ignored) {}
+ }
+ return 1;
+ }
+
+ @Override
+ public boolean execute(String sql) throws SQLException {
+ executeQuery(sql);
+ return true;
+ }
+
+
+
+ @Override
+ public void close() throws SQLException {
+ if (currentResult != null) {
+ currentResult.close();
+ }
+ }
+
+ @Override
+ public int getMaxFieldSize() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public void setMaxFieldSize(int max) throws SQLException {
+
+ }
+
+ @Override
+ public int getMaxRows() throws SQLException {
+ return maxRows;
+ }
+
+ @Override
+ public void setMaxRows(int max) throws SQLException {
+ maxRows = max;
+ }
+
+ @Override
+ public void setEscapeProcessing(boolean enable) throws SQLException {
+
+ }
+
+ @Override
+ public int getQueryTimeout() throws SQLException {
+ return queryTimeout;
+ }
+
+ @Override
+ public void setQueryTimeout(int seconds) throws SQLException {
+ queryTimeout = seconds;
+ }
+
+ @Override
+ public void cancel() throws SQLException {
+
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+
+ }
+
+ @Override
+ public void setCursorName(String name) throws SQLException {
+
+ }
+
+ @Override
+ public ResultSet getResultSet() throws SQLException {
+ return currentResult;
+ }
+
+ @Override
+ public int getUpdateCount() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public boolean getMoreResults() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public void setFetchDirection(int direction) throws SQLException {
+
+ }
+
+ @Override
+ public int getFetchDirection() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public void setFetchSize(int rows) throws SQLException {
+
+ }
+
+ @Override
+ public int getFetchSize() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public int getResultSetConcurrency() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public int getResultSetType() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public void addBatch(String sql) throws SQLException {
+
+ }
+
+ @Override
+ public void clearBatch() throws SQLException {
+
+ }
+
+ @Override
+ public int[] executeBatch() throws SQLException {
+ return new int[0];
+ }
+
+ @Override
+ public Connection getConnection() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public boolean getMoreResults(int current) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public ResultSet getGeneratedKeys() throws SQLException {
+ return null;
+ }
+
+ @Override
+ public int executeUpdate(String sql, int autoGeneratedKeys) throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public int executeUpdate(String sql, int[] columnIndexes) throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public int executeUpdate(String sql, String[] columnNames) throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public boolean execute(String sql, int autoGeneratedKeys) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean execute(String sql, int[] columnIndexes) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public boolean execute(String sql, String[] columnNames) throws SQLException {
+ return false;
+ }
+
+ @Override
+ public int getResultSetHoldability() throws SQLException {
+ return 0;
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public void setPoolable(boolean poolable) throws SQLException {
+
+ }
+
+ @Override
+ public boolean isPoolable() throws SQLException {
+ return false;
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException {
+ return null;
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) throws SQLException {
+ return false;
+ }
+
+ public static String clickhousifySql(String sql) {
+ return clickhousifySql(sql, "TabSeparatedWithNamesAndTypes");
+ }
+
+ public static String clickhousifySql(String sql, String format) {
+ sql = sql.trim();
+ if (!sql.replace(";", "").trim().endsWith(" TabSeparatedWithNamesAndTypes")
+ && !sql.replace(";", "").trim().endsWith(" TabSeparated")
+ && !sql.replace(";", "").trim().endsWith(" JSONCompact")) {
+ if (sql.endsWith(";")) sql = sql.substring(0, sql.length() - 1);
+ sql += " FORMAT " + format + ';';
+ }
+ return sql;
+ }
+
+ private String extractTableName(String sql) {
+ String s = extractDBAndTableName(sql);
+ if (s.contains(".")) {
+ return s.substring(s.indexOf(".") + 1);
+ } else return s;
+ }
+
+ private String extractDBName(String sql) {
+ String s = extractDBAndTableName(sql);
+ if (s.contains(".")) {
+ return s.substring(0, s.indexOf("."));
+ } else {
+ return source.getDatabase();
+ }
+ }
+
+ private String extractDBAndTableName(String sql) {
+ // паршивый код, надо пиÑать или найти нормальный парÑер
+ if (CopypasteUtils.startsWithIgnoreCase(sql, "select")) {
+ String withoutStrings = CopypasteUtils.retainUnquoted(sql, '\'');
+ int fromIndex = withoutStrings.indexOf("from");
+ if (fromIndex == -1) fromIndex = withoutStrings.indexOf("FROM");
+ if (fromIndex != -1) {
+ String fromFrom = withoutStrings.substring(fromIndex);
+ String fromTable = fromFrom.substring("from".length()).trim();
+ return fromTable.split(" ")[0];
+ }
+ }
+ if (CopypasteUtils.startsWithIgnoreCase(sql, "desc")) {
+ return "system.columns"; // bullshit
+ }
+ if (CopypasteUtils.startsWithIgnoreCase(sql, "show")) {
+ return "system.tables"; // bullshit
+ }
+ return "system.unknown";
+ }
+
+ private InputStream getInputStream(String sql,
+ Map<String, String> additionalClickHouseDBParams,
+ boolean ignoreDatabase
+ ) throws CHException {
+ sql = clickhousifySql(sql);
+ log.debug("Executing SQL: " + sql);
+ URI uri = null;
+ try {
+ Map<String, String> params = getParams(ignoreDatabase);
+ if (additionalClickHouseDBParams != null && !additionalClickHouseDBParams.isEmpty()) {
+ params.putAll(additionalClickHouseDBParams);
+ }
+ List<String> paramPairs = new ArrayList<String>();
+ for (Map.Entry<String, String> entry : params.entrySet()) {
+ paramPairs.add(entry.getKey() + '=' + entry.getValue());
+ }
+ String query = CopypasteUtils.join(paramPairs, '&');
+ uri = new URI("http", null, source.getHost(), source.getPort(),
+ "/", query, null);
+ } catch (URISyntaxException e) {
+ log.error("Mailformed URL: " + e.getMessage());
+ throw new IllegalStateException("illegal configuration of db");
+ }
+ log.debug("Request url: " + uri);
+ HttpPost post = new HttpPost(uri);
+ post.setEntity(new StringEntity(sql, CopypasteUtils.UTF_8));
+ HttpEntity entity = null;
+ InputStream is = null;
+ try {
+ HttpResponse response = client.execute(post);
+ entity = response.getEntity();
+ if (response.getStatusLine().getStatusCode() != HttpURLConnection.HTTP_OK) {
+ String chMessage;
+ try {
+ InputStream messageStream = entity.getContent();
+ if (properties.isCompress()) {
+ messageStream = new ClickhouseLZ4Stream(messageStream);
+ }
+ chMessage = CopypasteUtils.toString(messageStream);
+ } catch (IOException e) {
+ chMessage = "error while read response "+ e.getMessage();
+ }
+ EntityUtils.consumeQuietly(entity);
+ throw ClickhouseExceptionSpecifier.specify(chMessage, source.getHost(), source.getPort());
+ }
+ if (entity.isStreaming()) {
+ is = entity.getContent();
+ } else {
+ FastByteArrayOutputStream baos = new FastByteArrayOutputStream();
+ entity.writeTo(baos);
+ is = baos.convertToInputStream();
+ }
+ return is;
+ } catch (IOException e) {
+ log.info("Error during connection to " + source + ", reporting failure to data source, message: " + e.getMessage());
+ EntityUtils.consumeQuietly(entity);
+ try { if (is != null) is.close(); } catch (IOException ignored) { }
+ log.info("Error sql: " + sql);
+ throw new CHException("Unknown IO exception", e);
+ }
+ }
+
+ public Map<String, String> getParams(boolean ignoreDatabase) {
+ Map<String, String> params = new HashMap<String, String>();
+ //в clickhouse бывают таблички без базы (т.е. в базе default)
+ if (!CopypasteUtils.isBlank(source.getDatabase()) && !ignoreDatabase) {
+ params.put("database", source.getDatabase());
+ }
+ if (properties.isCompress()) {
+ params.put("compress", "1");
+ }
+ // нам вÑегда нужны min и max в ответе
+ params.put("extremes", "1");
+ if (CopypasteUtils.isBlank(properties.getProfile())) {
+ if (properties.getMaxThreads() != null)
+ params.put("max_threads", String.valueOf(properties.getMaxThreads()));
+ // да, там в Ñекундах
+ params.put("max_execution_time", String.valueOf((properties.getSocketTimeout() + properties.getDataTransferTimeout()) / 1000));
+ if (properties.getMaxBlockSize() != null) {
+ params.put("max_block_size", String.valueOf(properties.getMaxBlockSize()));
+ }
+ } else {
+ params.put("profile", properties.getProfile());
+ }
+ //в ÐºÐ»Ð¸ÐºÑ…Ð°ÑƒÑ Ð¸Ð½Ð¾Ð³Ð´Ð° бывает user
+ if (properties.getUser() != null) {
+ params.put("user", properties.getUser());
+ }
+ return params;
+ }
+
+ @Override
+ public void closeOnCompletion() throws SQLException {
+ closeOnCompletion = true;
+ }
+
+ @Override
+ public boolean isCloseOnCompletion() throws SQLException {
+ return closeOnCompletion;
+ }
+}
diff --git a/src/main/java/ru/yandex/metrika/clickhouse/copypaste/ArrayToStringDeserializer.java b/src/main/java/ru/yandex/metrika/clickhouse/copypaste/ArrayToStringDeserializer.java
new file mode 100644
index 0000000000000000000000000000000000000000..aed75c7052452ee6c817dc8231001be69c640ff9
--- /dev/null
+++ b/src/main/java/ru/yandex/metrika/clickhouse/copypaste/ArrayToStringDeserializer.java
@@ -0,0 +1,64 @@
+package ru.yandex.metrika.clickhouse.copypaste;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.type.TypeFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.WeakHashMap;
+
+/**
+ * И вÑе маÑÑивы, прилетевшие из кликхауÑа, Ñтанут Ñтроками.
+ *
+ * @author lemmsh
+ * @since 7/25/14
+ */
+
+class ArrayToStringDeserializer extends JsonDeserializer<List<String>> {
+
+ // cache не concurrent, потому что ничего Ñтрашного, что поÑчитаем неÑколько раз
+ private static final Map<DeserializationContext, JsonDeserializer<Object>> deserializers = new WeakHashMap<DeserializationContext, JsonDeserializer<Object>>();
+
+ @Override
+ public List<String> deserialize(JsonParser jp, DeserializationContext ctxt) throws IOException {
+ JsonDeserializer<Object> deserializer = deserializers.get(ctxt);
+ if (deserializer == null) {
+ deserializer = ctxt.findContextualValueDeserializer(TypeFactory.defaultInstance()
+ .constructType(new TypeReference<List<Object>>() {
+ }), null);
+ deserializers.put(ctxt, deserializer);
+ }
+
+ final Object deserialized = deserializer.deserialize(jp, ctxt);
+ if (!(deserialized instanceof List)){
+ throw new IllegalStateException();
+ }
+ //noinspection unchecked
+ final List<Object> deserializedList = (List) deserialized;
+ List<String> result = new ArrayList<String>();
+ for (Object x : deserializedList) {
+ String v = null;
+ if (x != null) {
+ if (x instanceof List) {
+ try {
+ v = new ObjectMapper().writeValueAsString(x);
+ } catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ v = x.toString();
+ }
+ }
+ result.add(v);
+ }
+ return result;
+ }
+
+}
diff --git a/src/main/java/ru/yandex/metrika/clickhouse/copypaste/ClickhouseResponse.java b/src/main/java/ru/yandex/metrika/clickhouse/copypaste/ClickhouseResponse.java
new file mode 100644
index 0000000000000000000000000000000000000000..410b76f844060ea32052fee457663e65782a7c0c
--- /dev/null
+++ b/src/main/java/ru/yandex/metrika/clickhouse/copypaste/ClickhouseResponse.java
@@ -0,0 +1,131 @@
+package ru.yandex.metrika.clickhouse.copypaste;
+
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+
+import java.util.List;
+
+/**
+ * ÐœÑп-объект Ð´Ð»Ñ Ð´Ð¶ÐµÐºÑона Ð´Ð»Ñ Ð¾Ñ‚Ð²ÐµÑ‚Ð° кликхауÑа
+ * @author jkee
+ */
+public class ClickhouseResponse {
+ private List<Meta> meta;
+ @JsonDeserialize(contentUsing = ArrayToStringDeserializer.class)
+ private List<List<String>> data;
+ @JsonDeserialize(using = ArrayToStringDeserializer.class)
+ private List<String> totals;
+ private Extremes extremes;
+ private int rows;
+ private int rows_before_limit_at_least;
+
+
+ public static class Extremes {
+ @JsonDeserialize(using = ArrayToStringDeserializer.class)
+ private List<String> min;
+ @JsonDeserialize(using = ArrayToStringDeserializer.class)
+ private List<String> max;
+
+ public List<String> getMin() {
+ return min;
+ }
+
+ public void setMin(List<String> min) {
+ this.min = min;
+ }
+
+ public List<String> getMax() {
+ return max;
+ }
+
+ public void setMax(List<String> max) {
+ this.max = max;
+ }
+ }
+
+ public static class Meta {
+ private String name;
+ private String type;
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ @Override
+ public String toString() {
+ return "Meta{" +
+ "name='" + name + '\'' +
+ ", type='" + type + '\'' +
+ '}';
+ }
+ }
+
+ public Extremes getExtremes() {
+ return extremes;
+ }
+
+ public void setExtremes(Extremes extremes) {
+ this.extremes = extremes;
+ }
+
+ public List<Meta> getMeta() {
+ return meta;
+ }
+
+ public void setMeta(List<Meta> meta) {
+ this.meta = meta;
+ }
+
+ public List<List<String>> getData() {
+ return data;
+ }
+
+ public void setData(List<List<String>> data) {
+ this.data = data;
+ }
+
+ public int getRows() {
+ return rows;
+ }
+
+ public void setRows(int rows) {
+ this.rows = rows;
+ }
+
+ public int getRows_before_limit_at_least() {
+ return rows_before_limit_at_least;
+ }
+
+ public void setRows_before_limit_at_least(int rows_before_limit_at_least) { //TODO надо бы деÑериализацию подправить
+ this.rows_before_limit_at_least = rows_before_limit_at_least;
+ }
+
+ public List<String> getTotals() {
+ return totals;
+ }
+
+ public void setTotals(List<String> totals) {
+ this.totals = totals;
+ }
+
+ @Override
+ public String toString() {
+ return "ClickhouseResponse{" +
+ "meta=" + meta +
+ ", data=" + data +
+ ", rows=" + rows +
+ ", rows_before_limit_at_least=" + rows_before_limit_at_least +
+ '}';
+ }
+}
diff --git a/src/test/java/ru/yandex/metrika/clickhouse/CHPreparedStatementTest.java b/src/test/java/ru/yandex/metrika/clickhouse/CHPreparedStatementTest.java
index 49320b04a0f92ecf726e1b2c988e9a161136058b..ed8b6860049744601c9367aced527fa7e83b6225 100644
--- a/src/test/java/ru/yandex/metrika/clickhouse/CHPreparedStatementTest.java
+++ b/src/test/java/ru/yandex/metrika/clickhouse/CHPreparedStatementTest.java
@@ -15,18 +15,18 @@ public class CHPreparedStatementTest {
public void testParseSql() throws Exception {
assertEquals(new ArrayList<String>(){{
add("SELECT * FROM tbl");
- }}, CHPreparedStatement.parseSql("SELECT * FROM tbl"));
+ }}, CHPreparedStatementImpl.parseSql("SELECT * FROM tbl"));
assertEquals(new ArrayList<String>(){{
add("SELECT * FROM tbl WHERE t = ");
add("");
- }}, CHPreparedStatement.parseSql("SELECT * FROM tbl WHERE t = ?"));
+ }}, CHPreparedStatementImpl.parseSql("SELECT * FROM tbl WHERE t = ?"));
assertEquals(new ArrayList<String>(){{
add("SELECT 'a\\'\\\\sdfasdf?adsf\\\\' as `sadf\\`?` FROM tbl WHERE t = ");
add(" AND r = ");
add(" ORDER BY 1");
- }}, CHPreparedStatement.parseSql("SELECT 'a\\'\\\\sdfasdf?adsf\\\\' as `sadf\\`?` FROM tbl WHERE t = ? AND r = ? ORDER BY 1"));
+ }}, CHPreparedStatementImpl.parseSql("SELECT 'a\\'\\\\sdfasdf?adsf\\\\' as `sadf\\`?` FROM tbl WHERE t = ? AND r = ? ORDER BY 1"));
}
}
\ No newline at end of file