es 7.17.24 support

Author: shi-yuan

.travis.yml

@@ -4,7 +4,7 @@ jdk:
   - openjdk8
 
 before_install:
-  - wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.23-amd64.deb && sudo dpkg -i --force-confnew elasticsearch-7.17.23-amd64.deb
+  - wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.17.24-amd64.deb && sudo dpkg -i --force-confnew elasticsearch-7.17.24-amd64.deb
   - sudo cp ./src/test/resources/elasticsearch.yml /etc/elasticsearch/elasticsearch.yml
   - sudo cat /etc/elasticsearch/elasticsearch.yml
   - sudo java -version

pom.xml

@@ -3,7 +3,7 @@
 	<modelVersion>4.0.0</modelVersion>
 	<groupId>org.nlpcn</groupId>
 	<artifactId>elasticsearch-sql</artifactId>
-	<version>7.17.23.0</version>
+	<version>7.17.24.0</version>
 	<packaging>jar</packaging>
 	<description>Query elasticsearch using SQL</description>
 	<name>elasticsearch-sql</name>
@@ -44,7 +44,7 @@
 		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 		<runSuite>**/MainTestSuite.class</runSuite>
 		<elasticsearch.plugin.name>sql</elasticsearch.plugin.name>
-		<elasticsearch.version>7.17.23</elasticsearch.version>
+		<elasticsearch.version>7.17.24</elasticsearch.version>
 		<elasticsearch.plugin.classname>org.elasticsearch.plugin.nlpcn.SqlPlug</elasticsearch.plugin.classname>
         <druid.version>1.2.15</druid.version>
         <guava.version>30.1.1-jre</guava.version>

src/main/java/org/elasticsearch/plugin/nlpcn/ElasticJoinExecutor.java

@@ -49,14 +49,10 @@ protected ElasticJoinExecutor(JoinRequestBuilder requestBuilder) {
                             && (secondTableReturnedField == null || secondTableReturnedField.size() == 0);
     }
 
-    public void  sendResponse(RestChannel channel){
-        try {
-            XContentBuilder builder = ElasticUtils.hitsAsXContentBuilder(results,metaResults);
-            BytesRestResponse bytesRestResponse = new BytesRestResponse(RestStatus.OK, builder);
-            channel.sendResponse(bytesRestResponse);
-        } catch (IOException e) {
-            e.printStackTrace();
-        }
+    public void sendResponse(RestChannel channel) throws IOException {
+        XContentBuilder builder = ElasticUtils.hitsAsXContentBuilder(results, metaResults);
+        BytesRestResponse bytesRestResponse = new BytesRestResponse(RestStatus.OK, builder);
+        channel.sendResponse(bytesRestResponse);
     }
 
     @Override

src/main/java/org/elasticsearch/plugin/nlpcn/RestSqlAction.java

@@ -3,6 +3,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.plugin.nlpcn.executors.ActionRequestRestExecuterFactory;
@@ -12,18 +13,18 @@
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.RestStatus;
 import org.nlpcn.es4sql.SearchDao;
-import org.nlpcn.es4sql.exception.SqlParseException;
 import org.nlpcn.es4sql.query.QueryAction;
 
 import java.io.IOException;
-import java.sql.SQLFeatureNotSupportedException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ExecutorService;
 
 import static org.elasticsearch.rest.RestRequest.Method.GET;
 import static org.elasticsearch.rest.RestRequest.Method.POST;
@@ -55,27 +56,39 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
             // LOGGER.warn("Please use json format params, like: {\"sql\":\"SELECT * FROM test\"}");
         }
 
-        String sql = request.param("sql");
+        String sql = Optional.ofNullable(request.param("sql")).orElseGet(() -> request.content().utf8ToString());
+        boolean useThreadPool = request.paramAsBoolean("useThreadPool", false);
 
-        if (sql == null) {
-            sql = request.content().utf8ToString();
+        if (useThreadPool) {
+            ExecutorService executor = client.threadPool().executor("nlpcn_sql");
+            return channel -> executor.execute(() -> doSqlRequest(request, client, sql, channel));
         }
+        return channel -> doSqlRequest(request, client, sql, channel);
+    }
+
+    @Override
+    protected Set<String> responseParams() {
+        Set<String> responseParams = new HashSet<>(super.responseParams());
+        responseParams.addAll(Arrays.asList("sql", "flat", "separator", "_score", "_type", "_id", "_scroll_id", "newLine", "format", "showHeader", "quote", "useThreadPool"));
+        return Collections.unmodifiableSet(responseParams);
+    }
+
+    private void doSqlRequest(RestRequest request, NodeClient client, String sql, RestChannel channel) {
         try {
             SearchDao searchDao = new SearchDao(client);
-            QueryAction queryAction = null;
 
-            queryAction = searchDao.explain(sql);//zhongshu-comment 语法解析，将sql字符串解析为一个Java查询对象
+            //zhongshu-comment 语法解析，将sql字符串解析为一个Java查询对象
+            QueryAction queryAction = searchDao.explain(sql);
 
             // TODO add unit tests to explain. (rest level?)
             if (request.path().endsWith("/explain")) {
                 final String jsonExplanation = queryAction.explain().explain();
-                return channel -> channel.sendResponse(new BytesRestResponse(RestStatus.OK, XContentType.JSON.mediaType(), jsonExplanation));
+                channel.sendResponse(new BytesRestResponse(RestStatus.OK, XContentType.JSON.mediaType(), jsonExplanation));
             } else {
                 Map<String, String> params = request.params();
 
                 //zhongshu-comment 生成一个负责用rest方式查询es的对象RestExecutor，返回的实现类是：ElasticDefaultRestExecutor
                 RestExecutor restExecutor = ActionRequestRestExecuterFactory.createExecutor(params.get("format"));
-                final QueryAction finalQueryAction = queryAction;
                 //doing this hack because elasticsearch throws exception for un-consumed props
                 Map<String, String> additionalParams = new HashMap<>();
                 for (String paramName : responseParams()) {
@@ -86,19 +99,15 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
                 //zhongshu-comment restExecutor.execute()方法里会调用es查询的相关rest api
                 //zhongshu-comment restExecutor.execute()方法的第1、4个参数是框架传进来的参数，第2、3个参数是可以自己生成的参数，所以要多注重一点
                 //zhongshu-comment 默认调用的是ElasticDefaultRestExecutor这个子类
-                //todo 这是什么语法：搜索java8 -> lambda表达式：https://blog.csdn.net/ioriogami/article/details/12782141
-                return channel -> restExecutor.execute(client, additionalParams, finalQueryAction, channel);
+                restExecutor.execute(client, additionalParams, queryAction, channel);
+            }
+        } catch (Exception e) {
+            try {
+                channel.sendResponse(new BytesRestResponse(channel, e));
+            } catch (Exception inner) {
+                inner.addSuppressed(e);
+                LOGGER.error("failed to send failure response", inner);
             }
-        } catch (SqlParseException | SQLFeatureNotSupportedException e) {
-            e.printStackTrace();
         }
-        return null;
-    }
-
-    @Override
-    protected Set<String> responseParams() {
-        Set<String> responseParams = new HashSet<>(super.responseParams());
-        responseParams.addAll(Arrays.asList("sql", "flat", "separator", "_score", "_type", "_id", "_scroll_id", "newLine", "format", "showHeader", "quote"));
-        return Collections.unmodifiableSet(responseParams);
     }
 }

src/main/java/org/elasticsearch/plugin/nlpcn/SqlPlug.java

@@ -12,6 +12,8 @@
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestHandler;
+import org.elasticsearch.threadpool.ExecutorBuilder;
+import org.elasticsearch.threadpool.FixedExecutorBuilder;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -47,4 +49,9 @@ protected void configure() {
             }
         });
     }
+
+    @Override
+    public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settings) {
+        return Collections.singletonList(new FixedExecutorBuilder(settings, "nlpcn_sql", 10, 100, "thread_pool.nlpcn_sql"));
+    }
 }

src/main/java/org/elasticsearch/plugin/nlpcn/executors/ElasticDefaultRestExecutor.java

@@ -107,13 +107,9 @@ public String execute(Client client, Map<String, String> params, QueryAction que
 
     }
 
-    private void sendDefaultResponse(SearchHits hits, RestChannel channel) {
-        try {
-            XContentBuilder builder = ElasticUtils.hitsAsXContentBuilder(hits, new MetaSearchResult());
-            BytesRestResponse bytesRestResponse = new BytesRestResponse(RestStatus.OK, builder);
-            channel.sendResponse(bytesRestResponse);
-        } catch (IOException e) {
-            e.printStackTrace();
-        }
+    private void sendDefaultResponse(SearchHits hits, RestChannel channel) throws IOException {
+        XContentBuilder builder = ElasticUtils.hitsAsXContentBuilder(hits, new MetaSearchResult());
+        BytesRestResponse bytesRestResponse = new BytesRestResponse(RestStatus.OK, builder);
+        channel.sendResponse(bytesRestResponse);
     }
 }