Create pit service layer changes

server/src/internalClusterTest/java/org/opensearch/search/searchafter/SearchAfterIT.java

@@ -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;
@@ -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();

server/src/internalClusterTest/java/org/opensearch/search/slice/SearchSliceIT.java

@@ -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;
@@ -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;
 
@@ -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();
@@ -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);
@@ -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) {

server/src/main/java/org/opensearch/action/ActionModule.java

@@ -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;
@@ -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());
     }
 

server/src/main/java/org/opensearch/action/search/CreatePitAction.java

@@ -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/create";
+
+    private CreatePitAction() {
+        super(NAME, CreatePitResponse::new);
+    }
+}