Skip to content

Commit 7815c96

Browse files
manasvinibsrupal-bq
manasvinibs
and
rupal-bq
authored
Merge feature/pit branch to main (#2936)
* Add pit for join queries (#2703) * Add search after for join Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Enable search after by default Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add pit Signed-off-by: Rupal Mahajan <maharup@amazon.com> * nit Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Fix tests Signed-off-by: Rupal Mahajan <maharup@amazon.com> * ignore joinWithGeoIntersectNL Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Rerun CI with scroll Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Remove unused code and retrigger CI with search_after true Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Address comments Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Remove unused code change Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Update pit keep alive time with SQL_CURSOR_KEEP_ALIVE Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Fix scroll condition Signed-off-by: Rupal Mahajan <maharup@amazon.com> * nit Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add pit before query execution Signed-off-by: Rupal Mahajan <maharup@amazon.com> * nit Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Move pit from join request builder to executor Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Remove unused methods Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add pit in parent class's run() Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add comment for fetching subsequent result in NestedLoopsElasticExecutor Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Update comment Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add javadoc for pit handler Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add pit interface Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add pit handler unit test Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Fix failed unit test CI Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Fix spotless error Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Rename pit class and add logs Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Fix pit delete unit test Signed-off-by: Rupal Mahajan <maharup@amazon.com> --------- Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add pit for multi query (#2753) * Add search after for join Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Enable search after by default Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add pit Signed-off-by: Rupal Mahajan <maharup@amazon.com> * nit Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Fix tests Signed-off-by: Rupal Mahajan <maharup@amazon.com> * ignore joinWithGeoIntersectNL Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Rerun CI with scroll Signed-off-by: Rupal Mahajan <maharup@amazon.com> * draft Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Remove unused code and retrigger CI with search_after true Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Address comments Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Remove unused code change Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Update pit keep alive time with SQL_CURSOR_KEEP_ALIVE Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Fix scroll condition Signed-off-by: Rupal Mahajan <maharup@amazon.com> * nit Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add pit before query execution Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Refactor get response with pit method Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Update remaining scroll search calls Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Fix integ test failures Signed-off-by: Rupal Mahajan <maharup@amazon.com> * nit Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Move pit from join request builder to executor Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Remove unused methods Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Move pit from request to executor Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Fix pit.delete call missed while merge Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Move getResponseWithHits method to util class Signed-off-by: Rupal Mahajan <maharup@amazon.com> * add try catch for create delete pit in minus executor Signed-off-by: Rupal Mahajan <maharup@amazon.com> * move all common fields to ElasticHitsExecutor Signed-off-by: Rupal Mahajan <maharup@amazon.com> * add javadoc for ElasticHitsExecutor Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Add missing javadoc Signed-off-by: Rupal Mahajan <maharup@amazon.com> * Forcing an empty commit as last commit is stuck processing updates Signed-off-by: Rupal Mahajan <maharup@amazon.com> --------- Signed-off-by: Rupal Mahajan <maharup@amazon.com> --------- Signed-off-by: Rupal Mahajan <maharup@amazon.com> Signed-off-by: Manasvini B S <manasvis@amazon.com> Co-authored-by: Rupal Mahajan <maharup@amazon.com>
1 parent 05c961e commit 7815c96

15 files changed

Lines changed: 557 additions & 158 deletions

File tree

‎common/src/main/java/org/opensearch/sql/common/setting/Settings.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ public enum Key {
2323
SQL_SLOWLOG("plugins.sql.slowlog"),
2424
SQL_CURSOR_KEEP_ALIVE("plugins.sql.cursor.keep_alive"),
2525
SQL_DELETE_ENABLED("plugins.sql.delete.enabled"),
26+
SQL_PAGINATION_API_SEARCH_AFTER("plugins.sql.pagination.api"),
2627

2728
/** PPL Settings. */
2829
PPL_ENABLED("plugins.ppl.enabled"),

‎docs/dev/opensearch-pagination.md‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -477,4 +477,44 @@ Response:
477477
}
478478
479479
480+
```
481+
482+
#### plugins.sql.pagination.api
483+
484+
This setting controls whether the SQL search queries in OpenSearch use Point-In-Time (PIT) with search_after or the traditional scroll mechanism for fetching paginated results.
485+
486+
- Default Value: true
487+
- Possible Values: true or false
488+
- When set to true, the search query in the background uses PIT with search_after instead of scroll to retrieve paginated results. The Cursor Id returned to the user will encode relevant pagination query-related information, which will be used to fetch the subsequent pages of results.
489+
- This setting is node-level.
490+
- This setting can be updated dynamically.
491+
492+
Example:
493+
494+
```
495+
>> curl -H 'Content-Type: application/json' -X PUT localhost:9200/_cluster/settings -d '{
496+
"transient" : {
497+
"plugins.sql.pagination.api" : "true"
498+
}
499+
}'
500+
```
501+
502+
Response:
503+
504+
```
505+
{
506+
"acknowledged" : true,
507+
"persistent" : { },
508+
"transient" : {
509+
"plugins" : {
510+
"sql" : {
511+
"pagination" : {
512+
"api" : "true"
513+
}
514+
}
515+
}
516+
}
517+
}
518+
519+
480520
```

‎docs/user/admin/settings.rst‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -196,6 +196,50 @@ Result set::
196196

197197
Note: the legacy settings of ``opendistro.sql.cursor.keep_alive`` is deprecated, it will fallback to the new settings if you request an update with the legacy name.
198198

199+
plugins.sql.pagination.api
200+
================================
201+
202+
Description
203+
-----------
204+
205+
This setting controls whether the SQL search queries in OpenSearch use Point-In-Time (PIT) with search_after or the traditional scroll mechanism for fetching paginated results.
206+
207+
1. Default Value: true
208+
2. Possible Values: true or false
209+
3. When set to true, the search query in the background uses PIT with search_after instead of scroll to retrieve paginated results. The Cursor Id returned to the user will encode relevant pagination query-related information, which will be used to fetch the subsequent pages of results.
210+
4. This setting is node-level.
211+
5. This setting can be updated dynamically.
212+
213+
214+
Example
215+
-------
216+
217+
You can update the setting with a new value like this.
218+
219+
SQL query::
220+
221+
>> curl -H 'Content-Type: application/json' -X PUT localhost:9200/_plugins/_query/settings -d '{
222+
"transient" : {
223+
"plugins.sql.pagination.api" : "true"
224+
}
225+
}'
226+
227+
Result set::
228+
229+
{
230+
"acknowledged" : true,
231+
"persistent" : { },
232+
"transient" : {
233+
"plugins" : {
234+
"sql" : {
235+
"pagination" : {
236+
"api" : "true"
237+
}
238+
}
239+
}
240+
}
241+
}
242+
199243
plugins.query.size_limit
200244
===========================
201245

‎integ-test/src/test/java/org/opensearch/sql/legacy/JoinIT.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -288,6 +288,8 @@ public void hintMultiSearchCanRunFewTimesNL() throws IOException {
288288
Assert.assertThat(hits.length(), equalTo(42));
289289
}
290290

291+
// TODO: Fix joinWithGeoIntersectNL test when SQL_PAGINATION_API_SEARCH_AFTER is true
292+
@Ignore
291293
@Test
292294
public void joinWithGeoIntersectNL() throws IOException {
293295

@@ -455,7 +457,7 @@ public void joinParseCheckSelectedFieldsSplitNLConditionOrderGT() throws IOExcep
455457
"SELECT /*! USE_NL*/ a.firstname, a.lastname, a.gender, d.firstname, d.age FROM %s a"
456458
+ " JOIN %s d on a.age < d.age WHERE (d.firstname = 'Lynn' OR d.firstname ="
457459
+ " 'Obrien') AND a.firstname = 'Mcgee'",
458-
TEST_INDEX_PEOPLE,
460+
TEST_INDEX_PEOPLE2,
459461
TEST_INDEX_ACCOUNT);
460462

461463
JSONObject result = executeQuery(query);
@@ -501,7 +503,7 @@ public void joinParseCheckSelectedFieldsSplitNLConditionOrderLT() throws IOExcep
501503
"SELECT /*! USE_NL*/ a.firstname, a.lastname, a.gender, d.firstname, d.age FROM %s a"
502504
+ " JOIN %s d on a.age > d.age WHERE (d.firstname = 'Sandoval' OR d.firstname ="
503505
+ " 'Hewitt') AND a.firstname = 'Fulton'",
504-
TEST_INDEX_PEOPLE,
506+
TEST_INDEX_PEOPLE2,
505507
TEST_INDEX_ACCOUNT);
506508

507509
JSONObject result = executeQuery(query);

‎legacy/src/main/java/org/opensearch/sql/legacy/executor/ElasticHitsExecutor.java‎

Lines changed: 87 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,96 @@
55

66
package org.opensearch.sql.legacy.executor;
77

8+
import static org.opensearch.search.sort.FieldSortBuilder.DOC_FIELD_NAME;
9+
import static org.opensearch.search.sort.SortOrder.ASC;
10+
import static org.opensearch.sql.common.setting.Settings.Key.SQL_CURSOR_KEEP_ALIVE;
11+
import static org.opensearch.sql.common.setting.Settings.Key.SQL_PAGINATION_API_SEARCH_AFTER;
12+
813
import java.io.IOException;
14+
import org.apache.logging.log4j.LogManager;
15+
import org.apache.logging.log4j.Logger;
16+
import org.opensearch.action.search.SearchRequestBuilder;
17+
import org.opensearch.action.search.SearchResponse;
18+
import org.opensearch.client.Client;
19+
import org.opensearch.common.unit.TimeValue;
920
import org.opensearch.search.SearchHits;
21+
import org.opensearch.search.builder.PointInTimeBuilder;
22+
import org.opensearch.sql.legacy.domain.Select;
23+
import org.opensearch.sql.legacy.esdomain.LocalClusterState;
1024
import org.opensearch.sql.legacy.exception.SqlParseException;
25+
import org.opensearch.sql.legacy.pit.PointInTimeHandler;
26+
27+
/** Executor for search requests with pagination. */
28+
public abstract class ElasticHitsExecutor {
29+
protected static final Logger LOG = LogManager.getLogger();
30+
protected PointInTimeHandler pit;
31+
protected Client client;
32+
33+
/**
34+
* Executes search request
35+
*
36+
* @throws IOException If an input or output exception occurred
37+
* @throws SqlParseException If parsing exception occurred
38+
*/
39+
protected abstract void run() throws IOException, SqlParseException;
40+
41+
/**
42+
* Get search hits after execution
43+
*
44+
* @return Search hits
45+
*/
46+
protected abstract SearchHits getHits();
47+
48+
/**
49+
* Get response for search request with pit/scroll
50+
*
51+
* @param request search request
52+
* @param select sql select
53+
* @param size fetch size
54+
* @param previousResponse response for previous request
55+
* @param pit point in time
56+
* @return search response for subsequent request
57+
*/
58+
public SearchResponse getResponseWithHits(
59+
SearchRequestBuilder request,
60+
Select select,
61+
int size,
62+
SearchResponse previousResponse,
63+
PointInTimeHandler pit) {
64+
// Set Size
65+
request.setSize(size);
66+
SearchResponse responseWithHits;
1167

12-
/** Created by Eliran on 21/8/2016. */
13-
public interface ElasticHitsExecutor {
14-
void run() throws IOException, SqlParseException;
68+
if (LocalClusterState.state().getSettingValue(SQL_PAGINATION_API_SEARCH_AFTER)) {
69+
// Set sort field for search_after
70+
boolean ordered = select.isOrderdSelect();
71+
if (!ordered) {
72+
request.addSort(DOC_FIELD_NAME, ASC);
73+
}
74+
// Set PIT
75+
request.setPointInTime(new PointInTimeBuilder(pit.getPitId()));
76+
// from and size is alternate method to paginate result.
77+
// If select has from clause, search after is not required.
78+
if (previousResponse != null && select.getFrom().isEmpty()) {
79+
request.searchAfter(previousResponse.getHits().getSortFields());
80+
}
81+
responseWithHits = request.get();
82+
} else {
83+
// Set scroll
84+
TimeValue keepAlive = LocalClusterState.state().getSettingValue(SQL_CURSOR_KEEP_ALIVE);
85+
if (previousResponse != null) {
86+
responseWithHits =
87+
client
88+
.prepareSearchScroll(previousResponse.getScrollId())
89+
.setScroll(keepAlive)
90+
.execute()
91+
.actionGet();
92+
} else {
93+
request.setScroll(keepAlive);
94+
responseWithHits = request.get();
95+
}
96+
}
1597

16-
SearchHits getHits();
98+
return responseWithHits;
99+
}
17100
}

‎legacy/src/main/java/org/opensearch/sql/legacy/executor/join/ElasticJoinExecutor.java‎

Lines changed: 44 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -5,34 +5,33 @@
55

66
package org.opensearch.sql.legacy.executor.join;
77

8+
import static org.opensearch.sql.common.setting.Settings.Key.SQL_PAGINATION_API_SEARCH_AFTER;
9+
810
import java.io.IOException;
911
import java.util.Collection;
1012
import java.util.HashMap;
1113
import java.util.HashSet;
1214
import java.util.List;
1315
import java.util.Map;
1416
import java.util.Set;
15-
import org.apache.logging.log4j.LogManager;
16-
import org.apache.logging.log4j.Logger;
17+
import java.util.stream.Stream;
1718
import org.apache.lucene.search.TotalHits;
1819
import org.apache.lucene.search.TotalHits.Relation;
19-
import org.opensearch.action.search.SearchRequestBuilder;
2020
import org.opensearch.action.search.SearchResponse;
2121
import org.opensearch.client.Client;
2222
import org.opensearch.common.document.DocumentField;
23-
import org.opensearch.common.unit.TimeValue;
2423
import org.opensearch.core.rest.RestStatus;
2524
import org.opensearch.core.xcontent.XContentBuilder;
2625
import org.opensearch.index.mapper.MapperService;
2726
import org.opensearch.rest.BytesRestResponse;
2827
import org.opensearch.rest.RestChannel;
2928
import org.opensearch.search.SearchHit;
3029
import org.opensearch.search.SearchHits;
31-
import org.opensearch.search.sort.FieldSortBuilder;
32-
import org.opensearch.search.sort.SortOrder;
3330
import org.opensearch.sql.legacy.domain.Field;
31+
import org.opensearch.sql.legacy.esdomain.LocalClusterState;
3432
import org.opensearch.sql.legacy.exception.SqlParseException;
3533
import org.opensearch.sql.legacy.executor.ElasticHitsExecutor;
34+
import org.opensearch.sql.legacy.pit.PointInTimeHandlerImpl;
3635
import org.opensearch.sql.legacy.query.SqlElasticRequestBuilder;
3736
import org.opensearch.sql.legacy.query.join.HashJoinElasticRequestBuilder;
3837
import org.opensearch.sql.legacy.query.join.JoinRequestBuilder;
@@ -41,23 +40,25 @@
4140
import org.opensearch.sql.legacy.query.planner.HashJoinQueryPlanRequestBuilder;
4241

4342
/** Created by Eliran on 15/9/2015. */
44-
public abstract class ElasticJoinExecutor implements ElasticHitsExecutor {
45-
private static final Logger LOG = LogManager.getLogger();
43+
public abstract class ElasticJoinExecutor extends ElasticHitsExecutor {
4644

4745
protected List<SearchHit> results; // Keep list to avoid copy to new array in SearchHits
4846
protected MetaSearchResult metaResults;
4947
protected final int MAX_RESULTS_ON_ONE_FETCH = 10000;
5048
private Set<String> aliasesOnReturn;
5149
private boolean allFieldsReturn;
50+
protected String[] indices;
5251

53-
protected ElasticJoinExecutor(JoinRequestBuilder requestBuilder) {
52+
protected ElasticJoinExecutor(Client client, JoinRequestBuilder requestBuilder) {
5453
metaResults = new MetaSearchResult();
5554
aliasesOnReturn = new HashSet<>();
5655
List<Field> firstTableReturnedField = requestBuilder.getFirstTable().getReturnedFields();
5756
List<Field> secondTableReturnedField = requestBuilder.getSecondTable().getReturnedFields();
5857
allFieldsReturn =
5958
(firstTableReturnedField == null || firstTableReturnedField.size() == 0)
6059
&& (secondTableReturnedField == null || secondTableReturnedField.size() == 0);
60+
indices = getIndices(requestBuilder);
61+
this.client = client;
6162
}
6263

6364
public void sendResponse(RestChannel channel) throws IOException {
@@ -85,10 +86,22 @@ public void sendResponse(RestChannel channel) throws IOException {
8586
}
8687

8788
public void run() throws IOException, SqlParseException {
88-
long timeBefore = System.currentTimeMillis();
89-
results = innerRun();
90-
long joinTimeInMilli = System.currentTimeMillis() - timeBefore;
91-
this.metaResults.setTookImMilli(joinTimeInMilli);
89+
try {
90+
long timeBefore = System.currentTimeMillis();
91+
if (LocalClusterState.state().getSettingValue(SQL_PAGINATION_API_SEARCH_AFTER)) {
92+
pit = new PointInTimeHandlerImpl(client, indices);
93+
pit.create();
94+
}
95+
results = innerRun();
96+
long joinTimeInMilli = System.currentTimeMillis() - timeBefore;
97+
this.metaResults.setTookImMilli(joinTimeInMilli);
98+
} catch (Exception e) {
99+
LOG.error("Failed during join query run.", e);
100+
} finally {
101+
if (LocalClusterState.state().getSettingValue(SQL_PAGINATION_API_SEARCH_AFTER)) {
102+
pit.delete();
103+
}
104+
}
92105
}
93106

94107
protected abstract List<SearchHit> innerRun() throws IOException, SqlParseException;
@@ -103,7 +116,7 @@ public SearchHits getHits() {
103116
public static ElasticJoinExecutor createJoinExecutor(
104117
Client client, SqlElasticRequestBuilder requestBuilder) {
105118
if (requestBuilder instanceof HashJoinQueryPlanRequestBuilder) {
106-
return new QueryPlanElasticExecutor((HashJoinQueryPlanRequestBuilder) requestBuilder);
119+
return new QueryPlanElasticExecutor(client, (HashJoinQueryPlanRequestBuilder) requestBuilder);
107120
} else if (requestBuilder instanceof HashJoinElasticRequestBuilder) {
108121
HashJoinElasticRequestBuilder hashJoin = (HashJoinElasticRequestBuilder) requestBuilder;
109122
return new HashJoinElasticExecutor(client, hashJoin);
@@ -256,23 +269,22 @@ protected void updateMetaSearchResults(SearchResponse searchResponse) {
256269
this.metaResults.updateTimeOut(searchResponse.isTimedOut());
257270
}
258271

259-
protected SearchResponse scrollOneTimeWithMax(
260-
Client client, TableInJoinRequestBuilder tableRequest) {
261-
SearchRequestBuilder scrollRequest =
262-
tableRequest
263-
.getRequestBuilder()
264-
.setScroll(new TimeValue(60000))
265-
.setSize(MAX_RESULTS_ON_ONE_FETCH);
266-
boolean ordered = tableRequest.getOriginalSelect().isOrderdSelect();
267-
if (!ordered) {
268-
scrollRequest.addSort(FieldSortBuilder.DOC_FIELD_NAME, SortOrder.ASC);
269-
}
270-
SearchResponse responseWithHits = scrollRequest.get();
271-
// on ordered select - not using SCAN , elastic returns hits on first scroll
272-
// es5.0 elastic always return docs on scan
273-
// if(!ordered)
274-
// responseWithHits = client.prepareSearchScroll(responseWithHits.getScrollId())
275-
// .setScroll(new TimeValue(600000)).get();
276-
return responseWithHits;
272+
public SearchResponse getResponseWithHits(
273+
TableInJoinRequestBuilder tableRequest, int size, SearchResponse previousResponse) {
274+
275+
return getResponseWithHits(
276+
tableRequest.getRequestBuilder(),
277+
tableRequest.getOriginalSelect(),
278+
size,
279+
previousResponse,
280+
pit);
281+
}
282+
283+
public String[] getIndices(JoinRequestBuilder joinRequestBuilder) {
284+
return Stream.concat(
285+
Stream.of(joinRequestBuilder.getFirstTable().getOriginalSelect().getIndexArr()),
286+
Stream.of(joinRequestBuilder.getSecondTable().getOriginalSelect().getIndexArr()))
287+
.distinct()
288+
.toArray(String[]::new);
277289
}
278290
}

0 commit comments

Comments
 (0)