x

src/main/java/org/elasticsearch/jdbc/AbstractFeatureNotSupportedPreparedStatement.java

@@ -0,0 +1,186 @@
+package org.elasticsearch.jdbc;
+
+
+import java.io.InputStream;
+import java.io.Reader;
+import java.net.URL;
+import java.sql.*;
+
+public abstract class AbstractFeatureNotSupportedPreparedStatement extends ElasticStatement implements PreparedStatement {
+
+    public AbstractFeatureNotSupportedPreparedStatement(ElasticConnection connection) {
+        super(connection);
+    }
+
+    @Override
+    public ResultSet executeQuery() throws SQLException {
+        throw new SQLFeatureNotSupportedException("executeQuery");
+    }
+
+    @Override
+    public int executeUpdate() throws SQLException {
+        throw new SQLFeatureNotSupportedException("executeQuery");
+    }
+
+    @Override
+    public void setNull(int parameterIndex, int sqlType) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setNull");
+    }
+
+
+    @Override
+    public void setBytes(int parameterIndex, byte[] x) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setBytes");
+    }
+
+    @Override
+    public void setAsciiStream(int parameterIndex, InputStream x, int length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setAsciiStream");
+    }
+
+    @Override
+    public void setUnicodeStream(int parameterIndex, InputStream x, int length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setUnicodeStream");
+    }
+
+    @Override
+    public void setBinaryStream(int parameterIndex, InputStream x, int length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setBinaryStream");
+    }
+
+
+    @Override
+    public void setObject(int parameterIndex, Object x, int targetSqlType) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setObject");
+    }
+
+    @Override
+    public void addBatch() throws SQLException {
+        throw new SQLFeatureNotSupportedException("addBatch");
+    }
+
+    @Override
+    public void setCharacterStream(int parameterIndex, Reader reader, int length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setCharacterStream");
+    }
+
+    @Override
+    public void setRef(int parameterIndex, Ref x) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setRef");
+    }
+
+    @Override
+    public void setBlob(int parameterIndex, Blob x) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setRef");
+    }
+
+    @Override
+    public void setClob(int parameterIndex, Clob x) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setRef");
+    }
+
+    @Override
+    public void setArray(int parameterIndex, Array x) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setArray");
+    }
+
+    @Override
+    public void setURL(int parameterIndex, URL x) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setURL");
+    }
+
+    @Override
+    public void setRowId(int parameterIndex, RowId x) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setRowId");
+    }
+
+    @Override
+    public void setNString(int parameterIndex, String value) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setNString");
+    }
+
+    @Override
+    public void setNCharacterStream(int parameterIndex, Reader value, long length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setNCharacterStream");
+    }
+
+    @Override
+    public void setNClob(int parameterIndex, NClob value) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setNClob");
+    }
+
+    @Override
+    public void setClob(int parameterIndex, Reader reader, long length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setClob");
+    }
+
+    @Override
+    public void setBlob(int parameterIndex, InputStream inputStream, long length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setBlob");
+    }
+
+    @Override
+    public void setNClob(int parameterIndex, Reader reader, long length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setNClob");
+    }
+
+    @Override
+    public void setSQLXML(int parameterIndex, SQLXML xmlObject) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setSQLXML");
+    }
+
+    @Override
+    public void setObject(int parameterIndex, Object x, int targetSqlType, int scaleOrLength) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setObject");
+    }
+
+    @Override
+    public void setAsciiStream(int parameterIndex, InputStream x, long length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setAsciiStream");
+    }
+
+    @Override
+    public void setBinaryStream(int parameterIndex, InputStream x, long length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setBinaryStream");
+    }
+
+    @Override
+    public void setCharacterStream(int parameterIndex, Reader reader, long length) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setCharacterStream");
+    }
+
+    @Override
+    public void setAsciiStream(int parameterIndex, InputStream x) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setAsciiStream");
+    }
+
+    @Override
+    public void setBinaryStream(int parameterIndex, InputStream x) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setBinaryStream");
+    }
+
+    @Override
+    public void setCharacterStream(int parameterIndex, Reader reader) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setCharacterStream");
+    }
+
+    @Override
+    public void setNCharacterStream(int parameterIndex, Reader value) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setNCharacterStream");
+    }
+
+    @Override
+    public void setClob(int parameterIndex, Reader reader) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setClob");
+    }
+
+    @Override
+    public void setBlob(int parameterIndex, InputStream inputStream) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setBlob");
+    }
+
+    @Override
+    public void setNClob(int parameterIndex, Reader reader) throws SQLException {
+        throw new SQLFeatureNotSupportedException("setNClob");
+    }
+}

src/main/java/org/elasticsearch/jdbc/AbstractPreparedStatement.java

@@ -0,0 +1,117 @@
+package org.elasticsearch.jdbc;
+
+import java.math.BigDecimal;
+import java.sql.*;
+import java.util.Calendar;
+
+public abstract class AbstractPreparedStatement extends AbstractFeatureNotSupportedPreparedStatement {
+
+    public AbstractPreparedStatement(ElasticConnection connection) {
+        super(connection);
+    }
+
+    @Override
+    public void setBoolean(int parameterIndex, boolean x) throws SQLException {
+
+    }
+
+    @Override
+    public void setByte(int parameterIndex, byte x) throws SQLException {
+
+    }
+
+    @Override
+    public void setShort(int parameterIndex, short x) throws SQLException {
+
+    }
+
+    @Override
+    public void setInt(int parameterIndex, int x) throws SQLException {
+
+    }
+
+    @Override
+    public void setLong(int parameterIndex, long x) throws SQLException {
+
+    }
+
+    @Override
+    public void setFloat(int parameterIndex, float x) throws SQLException {
+
+    }
+
+    @Override
+    public void setDouble(int parameterIndex, double x) throws SQLException {
+
+    }
+
+    @Override
+    public void setBigDecimal(int parameterIndex, BigDecimal x) throws SQLException {
+
+    }
+
+    @Override
+    public void setString(int parameterIndex, String x) throws SQLException {
+
+    }
+
+    @Override
+    public void setDate(int parameterIndex, Date x) throws SQLException {
+
+    }
+
+    @Override
+    public void setTime(int parameterIndex, Time x) throws SQLException {
+
+    }
+
+    @Override
+    public void setTimestamp(int parameterIndex, Timestamp x) throws SQLException {
+
+    }
+
+    @Override
+    public void clearParameters() throws SQLException {
+
+    }
+
+    @Override
+    public void setObject(int parameterIndex, Object x) throws SQLException {
+
+    }
+
+    @Override
+    public boolean execute() throws SQLException {
+        return false;
+    }
+
+    @Override
+    public ResultSetMetaData getMetaData() throws SQLException {
+        return new ElasticResultSetMetaData();
+    }
+
+    @Override
+    public void setDate(int parameterIndex, Date x, Calendar cal) throws SQLException {
+
+    }
+
+    @Override
+    public void setTime(int parameterIndex, Time x, Calendar cal) throws SQLException {
+
+    }
+
+    @Override
+    public void setTimestamp(int parameterIndex, Timestamp x, Calendar cal) throws SQLException {
+
+    }
+
+    @Override
+    public void setNull(int parameterIndex, int sqlType, String typeName) throws SQLException {
+
+    }
+
+    @Override
+    public ParameterMetaData getParameterMetaData() throws SQLException {
+        return null;
+    }
+}

src/main/java/org/elasticsearch/jdbc/ElasticStatement.java

@@ -1,13 +1,16 @@
 package org.elasticsearch.jdbc;
 
+import com.google.common.collect.Lists;
 import org.elasticsearch.action.search.SearchRequestBuilder;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.dsl.bean.ElasticSqlParseResult;
 import org.elasticsearch.dsl.parser.ElasticSql2DslParser;
+import org.elasticsearch.search.SearchHit;
 
 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.List;
 
 public class ElasticStatement extends AbstractStatement {
 
@@ -17,7 +20,7 @@ public class ElasticStatement extends AbstractStatement {
         this.connection = connection;
     }
 
-    private ResultSet executeResult;
+    protected ResultSet executeResult;
 
     @Override
     public ResultSet executeQuery(String sql) throws SQLException {
@@ -25,9 +28,22 @@ public class ElasticStatement extends AbstractStatement {
         ElasticSqlParseResult parseResult = sql2DslParser.parse(sql);
 
         SearchRequestBuilder searchRequest = parseResult.toRequest(connection.getClient());
-        SearchResponse searchResponse = searchRequest.execute().actionGet();
+        SearchResponse searchResponse = SearchActionExecutor.get().syncExecute(searchRequest);
 
-        return executeResult = new ElasticResultSet(this, searchResponse.toString());
+        SearchResponseGson searchResponseGson = new SearchResponseGson();
+        searchResponseGson.setFailedShards(searchResponse.getFailedShards());
+        searchResponseGson.setSuccessfulShards(searchResponse.getSuccessfulShards());
+        searchResponseGson.setTookInMillis(searchResponse.getTookInMillis());
+        searchResponseGson.setTotalShards(searchResponse.getTotalShards());
+        searchResponseGson.setTotalHits(searchResponse.getHits().getTotalHits());
+
+        List<String> hits = Lists.newLinkedList();
+        for (SearchHit searchHit : searchResponse.getHits().getHits()) {
+            hits.add(searchHit.getSourceAsString());
+        }
+        searchResponseGson.setDocList(hits);
+
+        return executeResult = new ElasticResultSet(this, searchResponseGson.toJson());
     }
 
     @Override

src/main/java/org/elasticsearch/jdbc/SearchActionExecutor.java

@@ -1,4 +1,38 @@
 package org.elasticsearch.jdbc;
 
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestBuilder;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ListenableActionFuture;
+
 public class SearchActionExecutor {
+
+    private static final SearchActionExecutor searchActionExecutor = new SearchActionExecutor();
+
+    private SearchActionExecutor() {
+
+    }
+
+    public static SearchActionExecutor get() {
+        return searchActionExecutor;
+    }
+
+    public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> Response syncExecuteWithException(RequestBuilder requestBuilder) {
+        ListenableActionFuture<Response> searchActionFuture = requestBuilder.execute();
+        return searchActionFuture.actionGet();
+    }
+
+    public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> Response syncExecute(RequestBuilder requestBuilder) {
+        try {
+            ListenableActionFuture<Response> searchActionFuture = requestBuilder.execute();
+            return searchActionFuture.actionGet();
+        }
+        catch (Exception ex) {
+            return null;
+        }
+    }
+
+    public <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void asyncExecute(RequestBuilder requestBuilder) {
+        requestBuilder.execute();
+    }
 }

src/main/java/org/elasticsearch/jdbc/SearchResponseGson.java

@@ -0,0 +1,82 @@
+package org.elasticsearch.jdbc;
+
+import com.google.common.collect.Lists;
+import com.google.common.reflect.TypeToken;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+
+import java.util.List;
+
+public class SearchResponseGson {
+    private int totalShards;
+    private int failedShards;
+    private int successfulShards;
+    private long tookInMillis;
+    private long totalHits;
+
+    private List<String> docList;
+
+
+    public int getTotalShards() {
+        return totalShards;
+    }
+
+    public long getTotalHits() {
+        return totalHits;
+    }
+
+    public void setTotalHits(long totalHits) {
+        this.totalHits = totalHits;
+    }
+
+    public void setTotalShards(int totalShards) {
+        this.totalShards = totalShards;
+    }
+
+    public int getFailedShards() {
+        return failedShards;
+    }
+
+    public void setFailedShards(int failedShards) {
+        this.failedShards = failedShards;
+    }
+
+    public int getSuccessfulShards() {
+        return successfulShards;
+    }
+
+    public void setSuccessfulShards(int successfulShards) {
+        this.successfulShards = successfulShards;
+    }
+
+    public long getTookInMillis() {
+        return tookInMillis;
+    }
+
+    public void setTookInMillis(long tookInMillis) {
+        this.tookInMillis = tookInMillis;
+    }
+
+    public <T> List<T> getDocList(TypeToken<T> typeToken) {
+        List<T> docs = Lists.newLinkedList();
+        Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ").create();
+        for (String docJson : docList) {
+            docs.add(gson.fromJson(docJson, typeToken.getType()));
+        }
+        return docs;
+    }
+
+    public void setDocList(List<String> docList) {
+        this.docList = docList;
+    }
+
+    @Override
+    public String toString() {
+        return toJson();
+    }
+
+    public String toJson() {
+        return new Gson().toJson(this, SearchResponseGson.class);
+    }
+
+}

src/main/java/org/elasticsearch/jdbc/TransportClientFactory.java

@@ -24,7 +24,7 @@ public class TransportClientFactory {
 
     public static TransportClient createTransportClientFromUrl(String url) {
         if (clientMap.containsKey(url)) {
-            return clientMap.get(url);
+            clientMap.remove(url);
         }
 
         Settings.Builder settingBuilder = Settings.settingsBuilder();

src/test/java/org/elasticsearch/jdbc/ElasticDriverTest.java

@@ -1,6 +1,8 @@
 package org.elasticsearch.jdbc;
 
 
+import com.google.common.reflect.TypeToken;
+import com.google.gson.Gson;
 import org.junit.Test;
 
 import java.sql.Connection;
@@ -8,6 +10,7 @@ import java.sql.Driver;
 import java.sql.DriverManager;
 import java.sql.ResultSet;
 import java.util.Enumeration;
+import java.util.List;
 
 public class ElasticDriverTest {
     private static final String driver = "org.elasticsearch.jdbc.ElasticDriver";
@@ -55,7 +58,27 @@ public class ElasticDriverTest {
 
         while(resultSet.next()) {
             String json = resultSet.getString(1);
-            System.out.println(json);
+            SearchResponseGson searchResponse = new Gson().fromJson(json, SearchResponseGson.class);
+
+            System.out.println(searchResponse.getTookInMillis());
+
+            List<Lib> libList = searchResponse.getDocList(new TypeToken<Lib>(){});
+
+            for (Lib lib : libList) {
+                System.out.println(lib.getName());
+            }
         }
     }
 }
+
+class Lib {
+    private String name;
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+}