Skip to content

Commit c16b8ff

Browse files
shi-yuanshi-yuan
committed
support scroll_id in sql #575 
1 parent d51a681 commit c16b8ff

File tree

5 files changed

+40
-47
lines changed

5 files changed

+40
-47
lines changed

‎pom.xml

+1-1
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@
33
<modelVersion>4.0.0</modelVersion>
44
<groupId>org.nlpcn</groupId>
55
<artifactId>elasticsearch-sql</artifactId>
6-
<version>6.2.4.1</version>
6+
<version>6.2.4.2</version>
77
<packaging>jar</packaging>
88
<description>Query elasticsearch using SQL</description>
99
<name>elasticsearch-sql</name>

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

+4-1
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
77
import org.elasticsearch.action.search.SearchRequest;
88
import org.elasticsearch.action.search.SearchResponse;
9+
import org.elasticsearch.action.search.SearchScrollRequest;
910
import org.elasticsearch.client.Client;
1011
import org.elasticsearch.index.reindex.BulkIndexByScrollResponseContentListener;
1112
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
@@ -47,11 +48,13 @@ public void execute(Client client, Map<String, String> params, QueryAction query
4748
executor.run();
4849
sendDefaultResponse(executor.getHits(), channel);
4950
} else if (request instanceof SearchRequest) {
50-
client.search((SearchRequest) request, new RestStatusToXContentListener<SearchResponse>(channel));
51+
client.search((SearchRequest) request, new RestStatusToXContentListener<>(channel));
5152
} else if (request instanceof DeleteByQueryRequest) {
5253
requestBuilder.getBuilder().execute(new BulkIndexByScrollResponseContentListener(channel, Maps.newHashMap()));
5354
} else if (request instanceof GetIndexRequest) {
5455
requestBuilder.getBuilder().execute(new GetIndexRequestRestListener(channel, (GetIndexRequest) request));
56+
} else if (request instanceof SearchScrollRequest) {
57+
client.searchScroll((SearchScrollRequest) request, new RestStatusToXContentListener<>(channel));
5558
} else {
5659
throw new Exception(String.format("Unsupported ActionRequest provided: %s", request.getClass().getName()));
5760
}

‎src/main/java/org/nlpcn/es4sql/domain/hints/HintFactory.java

+10-8
Original file line numberDiff line numberDiff line change
@@ -60,15 +60,17 @@ public static Hint getHintFromString(String hintAsString) throws SqlParseExcepti
6060
int multiSearchSize = Integer.parseInt(number[0]);
6161
return new Hint(HintType.NL_MULTISEARCH_SIZE,new Object[]{multiSearchSize});
6262
}
63-
if(hintAsString.startsWith("! USE_SCROLL")){
64-
String[] scrollParams = getParamsFromHint(hintAsString,"! USE_SCROLL");
65-
int docsPerShardFetch = 50;
66-
int timeout = 60000;
67-
if(scrollParams != null && scrollParams.length ==2) {
68-
docsPerShardFetch = Integer.parseInt(scrollParams[0]);
69-
timeout = Integer.parseInt(scrollParams[1]);
63+
if (hintAsString.startsWith("! USE_SCROLL")) {
64+
String[] scrollParams = getParamsFromHint(hintAsString, "! USE_SCROLL");
65+
if (scrollParams != null && scrollParams.length == 2) {
66+
String param = scrollParams[0];
67+
return new Hint(HintType.USE_SCROLL,
68+
new Object[]{(param.startsWith("\"") && param.endsWith("\"")) || (param.startsWith("'") && param.endsWith("'")) ? param.substring(1, param.length() - 1) :
69+
Integer.parseInt(param),
70+
Integer.parseInt(scrollParams[1])});
71+
} else {
72+
return new Hint(HintType.USE_SCROLL, new Object[]{50, 60000});
7073
}
71-
return new Hint(HintType.USE_SCROLL, new Object[]{docsPerShardFetch,timeout});
7274
}
7375
if(hintAsString.startsWith("! IGNORE_UNAVAILABLE")){
7476
return new Hint(HintType.IGNORE_UNAVAILABLE,null);

‎src/main/java/org/nlpcn/es4sql/query/DefaultQueryAction.java

+22-27
Original file line numberDiff line numberDiff line change
@@ -4,9 +4,7 @@
44
import java.util.List;
55
import java.util.ArrayList;
66

7-
import org.elasticsearch.action.search.SearchAction;
8-
import org.elasticsearch.action.search.SearchRequestBuilder;
9-
import org.elasticsearch.action.search.SearchType;
7+
import org.elasticsearch.action.search.*;
108
import org.elasticsearch.client.Client;
119
import org.elasticsearch.common.unit.TimeValue;
1210
import org.elasticsearch.index.query.BoolQueryBuilder;
@@ -41,7 +39,18 @@ public void intialize(SearchRequestBuilder request) throws SqlParseException {
4139

4240
@Override
4341
public SqlElasticSearchRequestBuilder explain() throws SqlParseException {
44-
this.request = new SearchRequestBuilder(client, SearchAction.INSTANCE);
42+
Hint scrollHint = null;
43+
for (Hint hint : select.getHints()) {
44+
if (hint.getType() == HintType.USE_SCROLL) {
45+
scrollHint = hint;
46+
break;
47+
}
48+
}
49+
if (scrollHint != null && scrollHint.getParams()[0] instanceof String) {
50+
return new SqlElasticSearchRequestBuilder(new SearchScrollRequestBuilder(client, SearchScrollAction.INSTANCE, (String) scrollHint.getParams()[0]).setScroll(new TimeValue((Integer) scrollHint.getParams()[1])));
51+
}
52+
53+
this.request = new SearchRequestBuilder(client, SearchAction.INSTANCE);
4554
setIndicesAndTypes();
4655

4756
setFields(select.getFields());
@@ -50,11 +59,15 @@ public SqlElasticSearchRequestBuilder explain() throws SqlParseException {
5059
setSorts(select.getOrderBys());
5160
setLimit(select.getOffset(), select.getRowCount());
5261

53-
boolean usedScroll = useScrollIfNeeded(select.isOrderdSelect());
54-
if (!usedScroll) {
55-
request.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
56-
}
57-
updateRequestWithIndexAndRoutingOptions(select, request);
62+
//
63+
if (scrollHint != null) {
64+
if (!select.isOrderdSelect())
65+
request.addSort(FieldSortBuilder.DOC_FIELD_NAME, SortOrder.ASC);
66+
request.setSize((Integer) scrollHint.getParams()[0]).setScroll(new TimeValue((Integer) scrollHint.getParams()[1]));
67+
} else {
68+
request.setSearchType(SearchType.DFS_QUERY_THEN_FETCH);
69+
}
70+
updateRequestWithIndexAndRoutingOptions(select, request);
5871
updateRequestWithHighlight(select, request);
5972
updateRequestWithCollapse(select, request);
6073
updateRequestWithPostFilter(select, request);
@@ -63,24 +76,6 @@ public SqlElasticSearchRequestBuilder explain() throws SqlParseException {
6376
return sqlElasticRequestBuilder;
6477
}
6578

66-
private boolean useScrollIfNeeded(boolean existsOrderBy) {
67-
Hint scrollHint = null;
68-
for (Hint hint : select.getHints()) {
69-
if (hint.getType() == HintType.USE_SCROLL) {
70-
scrollHint = hint;
71-
break;
72-
}
73-
}
74-
if (scrollHint != null) {
75-
int scrollSize = (Integer) scrollHint.getParams()[0];
76-
int timeoutInMilli = (Integer) scrollHint.getParams()[1];
77-
if (!existsOrderBy)
78-
request.addSort(FieldSortBuilder.DOC_FIELD_NAME, SortOrder.ASC);
79-
request.setScroll(new TimeValue(timeoutInMilli)).setSize(scrollSize);
80-
}
81-
return scrollHint != null;
82-
}
83-
8479
/**
8580
* Set indices and types to the search request.
8681
*/

‎src/main/java/org/nlpcn/es4sql/query/SqlElasticSearchRequestBuilder.java

+3-10
Original file line numberDiff line numberDiff line change
@@ -3,21 +3,14 @@
33
import org.elasticsearch.action.ActionRequest;
44
import org.elasticsearch.action.ActionRequestBuilder;
55
import org.elasticsearch.action.ActionResponse;
6-
import org.elasticsearch.action.search.SearchRequestBuilder;
7-
import org.elasticsearch.common.xcontent.ToXContent;
8-
import org.elasticsearch.common.xcontent.XContentBuilder;
9-
import org.elasticsearch.common.xcontent.XContentFactory;
10-
import org.elasticsearch.common.xcontent.XContentType;
11-
12-
import java.io.IOException;
136

147
/**
158
* Created by Eliran on 19/8/2015.
169
*/
1710
public class SqlElasticSearchRequestBuilder implements SqlElasticRequestBuilder {
18-
SearchRequestBuilder requestBuilder;
11+
ActionRequestBuilder requestBuilder;
1912

20-
public SqlElasticSearchRequestBuilder(SearchRequestBuilder requestBuilder) {
13+
public SqlElasticSearchRequestBuilder(ActionRequestBuilder requestBuilder) {
2114
this.requestBuilder = requestBuilder;
2215
}
2316

@@ -42,7 +35,7 @@ public ActionRequestBuilder getBuilder() {
4235
}
4336

4437
@Override
45-
public String toString(){
38+
public String toString() {
4639
return this.requestBuilder.toString();
4740
}
4841
}

0 commit comments

Comments
 (0)