Skip to content

Commit

Permalink
Create pit service layer changes
Browse files Browse the repository at this point in the history 
Signed-off-by: Bharathwaj G <bharath78910@gmail.com>
  • Loading branch information
@bharath-techie
bharath-techie committed Jul 15, 2022
1 parent 0e96a87 commit fa9946b
Show file tree
Hide file tree
Showing 32 changed files with 3,111 additions and 25 deletions.
58 changes: 58 additions & 0 deletions server/src/internalClusterTest/java/org/opensearch/search/searchafter/SearchAfterIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,15 +32,21 @@

package org.opensearch.search.searchafter;

import org.opensearch.action.ActionFuture;
import org.opensearch.action.admin.indices.create.CreateIndexRequestBuilder;
import org.opensearch.action.index.IndexRequestBuilder;
import org.opensearch.action.search.CreatePitAction;
import org.opensearch.action.search.CreatePitRequest;
import org.opensearch.action.search.CreatePitResponse;
import org.opensearch.action.search.SearchPhaseExecutionException;
import org.opensearch.action.search.SearchRequestBuilder;
import org.opensearch.action.search.SearchResponse;
import org.opensearch.action.search.ShardSearchFailure;
import org.opensearch.common.UUIDs;
import org.opensearch.common.unit.TimeValue;
import org.opensearch.common.xcontent.XContentBuilder;
import org.opensearch.search.SearchHit;
import org.opensearch.search.builder.PointInTimeBuilder;
import org.opensearch.search.sort.SortOrder;
import org.opensearch.test.OpenSearchIntegTestCase;
import org.hamcrest.Matchers;
Expand Down Expand Up @@ -155,6 +161,58 @@ public void testsShouldFail() throws Exception {
}
}

public void testPitWithSearchAfter() throws Exception {
assertAcked(client().admin().indices().prepareCreate("test").setMapping("field1", "type=long", "field2", "type=keyword").get());
ensureGreen();
indexRandom(
true,
client().prepareIndex("test").setId("0").setSource("field1", 0),
client().prepareIndex("test").setId("1").setSource("field1", 100, "field2", "toto"),
client().prepareIndex("test").setId("2").setSource("field1", 101),
client().prepareIndex("test").setId("3").setSource("field1", 99)
);

CreatePitRequest request = new CreatePitRequest(TimeValue.timeValueDays(1), true);
request.setIndices(new String[] { "test" });
ActionFuture<CreatePitResponse> execute = client().execute(CreatePitAction.INSTANCE, request);
CreatePitResponse pitResponse = execute.get();
SearchResponse sr = client().prepareSearch()
.addSort("field1", SortOrder.ASC)
.setQuery(matchAllQuery())
.searchAfter(new Object[] { 99 })
.setPointInTime(new PointInTimeBuilder(pitResponse.getId()))
.get();
assertEquals(2, sr.getHits().getHits().length);
sr = client().prepareSearch()
.addSort("field1", SortOrder.ASC)
.setQuery(matchAllQuery())
.searchAfter(new Object[] { 100 })
.setPointInTime(new PointInTimeBuilder(pitResponse.getId()))
.get();
assertEquals(1, sr.getHits().getHits().length);
sr = client().prepareSearch()
.addSort("field1", SortOrder.ASC)
.setQuery(matchAllQuery())
.searchAfter(new Object[] { 0 })
.setPointInTime(new PointInTimeBuilder(pitResponse.getId()))
.get();
assertEquals(3, sr.getHits().getHits().length);
/**
* Add new data and assert PIT results remain the same and normal search results gets refreshed
*/
indexRandom(true, client().prepareIndex("test").setId("4").setSource("field1", 102));
sr = client().prepareSearch()
.addSort("field1", SortOrder.ASC)
.setQuery(matchAllQuery())
.searchAfter(new Object[] { 0 })
.setPointInTime(new PointInTimeBuilder(pitResponse.getId()))
.get();
assertEquals(3, sr.getHits().getHits().length);
sr = client().prepareSearch().addSort("field1", SortOrder.ASC).setQuery(matchAllQuery()).searchAfter(new Object[] { 0 }).get();
assertEquals(4, sr.getHits().getHits().length);
client().admin().indices().prepareDelete("test").get();
}

public void testWithNullStrings() throws InterruptedException {
assertAcked(client().admin().indices().prepareCreate("test").setMapping("field2", "type=keyword").get());
ensureGreen();
Expand Down
86 changes: 84 additions & 2 deletions server/src/internalClusterTest/java/org/opensearch/search/slice/SearchSliceIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,13 @@

package org.opensearch.search.slice;

import org.opensearch.action.ActionFuture;
import org.opensearch.action.admin.indices.alias.IndicesAliasesRequest;

import org.opensearch.action.index.IndexRequestBuilder;
import org.opensearch.action.search.CreatePitAction;
import org.opensearch.action.search.CreatePitRequest;
import org.opensearch.action.search.CreatePitResponse;
import org.opensearch.action.search.SearchPhaseExecutionException;
import org.opensearch.action.search.SearchRequestBuilder;
import org.opensearch.action.search.SearchResponse;
Expand All @@ -46,6 +50,7 @@
import org.opensearch.search.Scroll;
import org.opensearch.search.SearchException;
import org.opensearch.search.SearchHit;
import org.opensearch.search.builder.PointInTimeBuilder;
import org.opensearch.search.sort.SortBuilders;
import org.opensearch.test.OpenSearchIntegTestCase;

Expand Down Expand Up @@ -86,7 +91,12 @@ private void setupIndex(int numDocs, int numberOfShards) throws IOException, Exe
client().admin()
.indices()
.prepareCreate("test")
.setSettings(Settings.builder().put("number_of_shards", numberOfShards).put("index.max_slices_per_scroll", 10000))
.setSettings(
Settings.builder()
.put("number_of_shards", numberOfShards)
.put("index.max_slices_per_scroll", 10000)
.put("index.max_slices_per_pit", 10000)
)
.setMapping(mapping)
);
ensureGreen();
Expand Down Expand Up @@ -129,6 +139,78 @@ public void testSearchSort() throws Exception {
}
}

public void testSearchSortWithoutPitOrScroll() throws Exception {
int numShards = randomIntBetween(1, 7);
int numDocs = randomIntBetween(100, 1000);
setupIndex(numDocs, numShards);
int fetchSize = randomIntBetween(10, 100);
SearchRequestBuilder request = client().prepareSearch("test")
.setQuery(matchAllQuery())
.setSize(fetchSize)
.addSort(SortBuilders.fieldSort("_doc"));
SliceBuilder sliceBuilder = new SliceBuilder("_id", 0, 4);
SearchPhaseExecutionException ex = expectThrows(SearchPhaseExecutionException.class, () -> request.slice(sliceBuilder).get());
assertTrue(ex.getMessage().contains("all shards failed"));
}

public void testSearchSortWithPIT() throws Exception {
int numShards = randomIntBetween(1, 7);
int numDocs = randomIntBetween(100, 1000);
setupIndex(numDocs, numShards);
int max = randomIntBetween(2, numShards * 3);
CreatePitRequest pitRequest = new CreatePitRequest(TimeValue.timeValueDays(1), true);
pitRequest.setIndices(new String[] { "test" });
ActionFuture<CreatePitResponse> execute = client().execute(CreatePitAction.INSTANCE, pitRequest);
CreatePitResponse pitResponse = execute.get();
for (String field : new String[] { "_id", "random_int", "static_int" }) {
int fetchSize = randomIntBetween(10, 100);

// test _doc sort
SearchRequestBuilder request = client().prepareSearch("test")
.setQuery(matchAllQuery())
.setPointInTime(new PointInTimeBuilder(pitResponse.getId()))
.setSize(fetchSize)
.addSort(SortBuilders.fieldSort("_doc"));
assertSearchSlicesWithPIT(request, field, max, numDocs);

// test numeric sort
request = client().prepareSearch("test")
.setQuery(matchAllQuery())
.setPointInTime(new PointInTimeBuilder(pitResponse.getId()))
.setSize(fetchSize)
.addSort(SortBuilders.fieldSort("random_int"));
assertSearchSlicesWithPIT(request, field, max, numDocs);
}
client().admin().indices().prepareDelete("test").get();
}

private void assertSearchSlicesWithPIT(SearchRequestBuilder request, String field, int numSlice, int numDocs) {
int totalResults = 0;
List<String> keys = new ArrayList<>();
for (int id = 0; id < numSlice; id++) {
SliceBuilder sliceBuilder = new SliceBuilder(field, id, numSlice);
SearchResponse searchResponse = request.slice(sliceBuilder).setFrom(0).get();
totalResults += searchResponse.getHits().getHits().length;
int expectedSliceResults = (int) searchResponse.getHits().getTotalHits().value;
int numSliceResults = searchResponse.getHits().getHits().length;
for (SearchHit hit : searchResponse.getHits().getHits()) {
assertTrue(keys.add(hit.getId()));
}
while (searchResponse.getHits().getHits().length > 0) {
searchResponse = request.setFrom(numSliceResults).slice(sliceBuilder).get();
totalResults += searchResponse.getHits().getHits().length;
numSliceResults += searchResponse.getHits().getHits().length;
for (SearchHit hit : searchResponse.getHits().getHits()) {
assertTrue(keys.add(hit.getId()));
}
}
assertThat(numSliceResults, equalTo(expectedSliceResults));
}
assertThat(totalResults, equalTo(numDocs));
assertThat(keys.size(), equalTo(numDocs));
assertThat(new HashSet(keys).size(), equalTo(numDocs));
}

public void testWithPreferenceAndRoutings() throws Exception {
int numShards = 10;
int totalDocs = randomIntBetween(100, 1000);
Expand Down Expand Up @@ -217,7 +299,7 @@ public void testInvalidQuery() throws Exception {
);
Throwable rootCause = findRootCause(exc);
assertThat(rootCause.getClass(), equalTo(SearchException.class));
assertThat(rootCause.getMessage(), equalTo("`slice` cannot be used outside of a scroll context"));
assertThat(rootCause.getMessage(), equalTo("`slice` cannot be used outside of a scroll context or PIT context"));
}

private void assertSearchSlicesWithScroll(SearchRequestBuilder request, String field, int numSlice, int numDocs) {
Expand Down
5 changes: 5 additions & 0 deletions server/src/main/java/org/opensearch/action/ActionModule.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -232,10 +232,12 @@
import org.opensearch.action.main.MainAction;
import org.opensearch.action.main.TransportMainAction;
import org.opensearch.action.search.ClearScrollAction;
import org.opensearch.action.search.CreatePitAction;
import org.opensearch.action.search.MultiSearchAction;
import org.opensearch.action.search.SearchAction;
import org.opensearch.action.search.SearchScrollAction;
import org.opensearch.action.search.TransportClearScrollAction;
import org.opensearch.action.search.TransportCreatePitAction;
import org.opensearch.action.search.TransportMultiSearchAction;
import org.opensearch.action.search.TransportSearchAction;
import org.opensearch.action.search.TransportSearchScrollAction;
Expand Down Expand Up @@ -657,6 +659,9 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
actions.register(DeleteDanglingIndexAction.INSTANCE, TransportDeleteDanglingIndexAction.class);
actions.register(FindDanglingIndexAction.INSTANCE, TransportFindDanglingIndexAction.class);

// point in time actions
actions.register(CreatePitAction.INSTANCE, TransportCreatePitAction.class);

return unmodifiableMap(actions.getRegistry());
}

Expand Down
23 changes: 23 additions & 0 deletions server/src/main/java/org/opensearch/action/search/CreatePitAction.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
/*
* SPDX-License-Identifier: Apache-2.0
*
* The OpenSearch Contributors require contributions made to
* this file be licensed under the Apache-2.0 license or a
* compatible open source license.
*/

package org.opensearch.action.search;

import org.opensearch.action.ActionType;

/**
* Action type for creating PIT reader context
*/
public class CreatePitAction extends ActionType<CreatePitResponse> {
public static final CreatePitAction INSTANCE = new CreatePitAction();
public static final String NAME = "indices:data/read/point_in_time";

private CreatePitAction() {
super(NAME, CreatePitResponse::new);
}
}
Loading

0 comments on commit fa9946b

Please sign in to comment.