es 8.15.1 support

Author: shi-yuan

.travis.yml

@@ -5,7 +5,7 @@ jdk:
 
 before_install:
   - sudo rm -rf /var/lib/elasticsearch
-  - curl https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.15.0-amd64.deb -o elasticsearch.deb && sudo dpkg -i --force-confnew elasticsearch.deb
+  - curl https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-8.15.1-amd64.deb -o elasticsearch.deb && sudo dpkg -i --force-confnew elasticsearch.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>8.15.0.0</version>
+	<version>8.15.1.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>8.15.0</elasticsearch.version>
+		<elasticsearch.version>8.15.1</elasticsearch.version>
 		<elasticsearch.plugin.classname>org.elasticsearch.plugin.nlpcn.SqlPlug</elasticsearch.plugin.classname>
         <druid.version>1.2.15</druid.version>
         <guava.version>32.0.0-jre</guava.version>

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

@@ -5,7 +5,7 @@
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.rest.RestChannel;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 import org.nlpcn.es4sql.exception.SqlParseException;
 import org.nlpcn.es4sql.query.SqlElasticDeleteByQueryRequestBuilder;
 import org.nlpcn.es4sql.query.SqlElasticRequestBuilder;
@@ -39,7 +39,7 @@ public void execute() throws Exception {
             executeJoinRequestAndSendResponse();
         }
 		else if (request instanceof SearchRequest) {
-			client.search((SearchRequest) request, new RestChunkedToXContentListener<>(channel));
+			client.search((SearchRequest) request, new RestRefCountedChunkedToXContentListener<>(channel));
 		} else if (requestBuilder instanceof SqlElasticDeleteByQueryRequestBuilder) {
             throw new UnsupportedOperationException("currently not support delete on elastic 2.0.0");
         }

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);
-            RestResponse bytesRestResponse = new RestResponse(RestStatus.OK, builder);
-            channel.sendResponse(bytesRestResponse);
-        } catch (IOException e) {
-            e.printStackTrace();
-        }
+    public void sendResponse(RestChannel channel) throws IOException {
+        XContentBuilder builder = ElasticUtils.hitsAsXContentBuilder(results, metaResults);
+        RestResponse bytesRestResponse = new RestResponse(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.internal.node.NodeClient;
+import org.elasticsearch.rest.RestChannel;
 import org.elasticsearch.rest.RestResponse;
 import org.elasticsearch.xcontent.XContentParseException;
 import org.elasticsearch.xcontent.XContentParser;
@@ -13,18 +14,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;
@@ -56,27 +57,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 RestResponse(RestStatus.OK, XContentType.JSON.mediaType(), jsonExplanation));
+                channel.sendResponse(new RestResponse(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()) {
@@ -87,19 +100,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 RestResponse(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

@@ -7,14 +7,18 @@
 import org.elasticsearch.common.settings.IndexScopedSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.features.NodeFeature;
 import org.elasticsearch.plugins.ActionPlugin;
 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;
+import java.util.List;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
@@ -40,4 +44,9 @@ public Collection<?> createComponents(PluginServices services) {
     public Collection<RestHandler> getRestHandlers(Settings settings, NamedWriteableRegistry namedWriteableRegistry, RestController restController, ClusterSettings clusterSettings, IndexScopedSettings indexScopedSettings, SettingsFilter settingsFilter, IndexNameExpressionResolver indexNameExpressionResolver, Supplier<DiscoveryNodes> nodesInCluster, Predicate<NodeFeature> clusterSupportsFeature) {
         return Collections.singletonList(new RestSqlAction());
     }
+
+    @Override
+    public List<ExecutorBuilder<?>> getExecutorBuilders(Settings settings) {
+        return Collections.singletonList(new FixedExecutorBuilder(settings, "nlpcn_sql", 10, 100, "thread_pool.nlpcn_sql", EsExecutors.TaskTrackingConfig.DEFAULT));
+    }
 }

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

@@ -10,7 +10,7 @@
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.rest.RestResponse;
-import org.elasticsearch.rest.action.RestChunkedToXContentListener;
+import org.elasticsearch.rest.action.RestRefCountedChunkedToXContentListener;
 import org.elasticsearch.xcontent.XContentBuilder;
 import org.elasticsearch.reindex.BulkIndexByScrollResponseContentListener;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
@@ -63,7 +63,7 @@ public void execute(Client client, Map<String, String> params, QueryAction query
         } else if (request instanceof SearchRequest) {
             //zhongshu-comment 对应的QueryAction实现子类：DefaultQueryAction、AggregationQueryAction
             //zhongshu-comment 对应的SqlElasticRequestBuilder实现子类：SqlElasticSearchRequestBuilder
-            client.search((SearchRequest) request, new RestChunkedToXContentListener<>(channel));
+            client.search((SearchRequest) request, new RestRefCountedChunkedToXContentListener<>(channel));
         } else if (request instanceof DeleteByQueryRequest) {
             //zhongshu-comment 对应的QueryAction实现子类：DeleteQueryAction
             //zhongshu-comment 对应的SqlElasticRequestBuilder实现子类：SqlElasticDeleteByQueryRequestBuilder
@@ -73,7 +73,7 @@ public void execute(Client client, Map<String, String> params, QueryAction query
             //zhongshu-comment 对应的SqlElasticRequestBuilder实现子类：是一个匿名内部类，跳进去queryAction.explain()看
             requestBuilder.getBuilder().execute(new GetIndexRequestRestListener(channel, (GetIndexRequest) request));
         } else if (request instanceof SearchScrollRequest) {
-            client.searchScroll((SearchScrollRequest) request, new RestChunkedToXContentListener<>(channel));
+            client.searchScroll((SearchScrollRequest) request, new RestRefCountedChunkedToXContentListener<>(channel));
         } else {
             throw new Exception(String.format("Unsupported ActionRequest provided: %s", request.getClass().getName()));
         }
@@ -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());
-            RestResponse bytesRestResponse = new RestResponse(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());
+        RestResponse bytesRestResponse = new RestResponse(RestStatus.OK, builder);
+        channel.sendResponse(bytesRestResponse);
     }
 }