Create PIT API

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

@@ -32,15 +32,15 @@
 
 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.SearchPhaseExecutionException;
-import org.opensearch.action.search.SearchRequestBuilder;
-import org.opensearch.action.search.SearchResponse;
-import org.opensearch.action.search.ShardSearchFailure;
+import org.opensearch.action.search.*;
 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;
@@ -50,7 +50,6 @@
 import java.util.Comparator;
 import java.util.Collections;
 import java.util.Arrays;
-
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
 import static org.opensearch.common.xcontent.XContentFactory.jsonBuilder;
 import static org.opensearch.index.query.QueryBuilders.matchAllQuery;
@@ -155,6 +154,57 @@ 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);
+    }
+
     public void testWithNullStrings() throws InterruptedException {
         assertAcked(client().admin().indices().prepareCreate("test").setMapping("field2", "type=keyword").get());
         ensureGreen();

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

@@ -236,10 +236,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;
@@ -400,6 +402,7 @@
 import org.opensearch.rest.action.ingest.RestSimulatePipelineAction;
 import org.opensearch.rest.action.search.RestClearScrollAction;
 import org.opensearch.rest.action.search.RestCountAction;
+import org.opensearch.rest.action.search.RestCreatePITAction;
 import org.opensearch.rest.action.search.RestExplainAction;
 import org.opensearch.rest.action.search.RestMultiSearchAction;
 import org.opensearch.rest.action.search.RestSearchAction;
@@ -660,6 +663,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         actions.register(ImportDanglingIndexAction.INSTANCE, TransportImportDanglingIndexAction.class);
         actions.register(DeleteDanglingIndexAction.INSTANCE, TransportDeleteDanglingIndexAction.class);
         actions.register(FindDanglingIndexAction.INSTANCE, TransportFindDanglingIndexAction.class);
+        actions.register(CreatePITAction.INSTANCE, TransportCreatePITAction.class);
 
         return unmodifiableMap(actions.getRegistry());
     }
@@ -832,6 +836,9 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster) {
         registerHandler.accept(new RestRepositoriesAction());
         registerHandler.accept(new RestSnapshotAction());
         registerHandler.accept(new RestTemplatesAction());
+
+        // Point in time API
+        registerHandler.accept(new RestCreatePITAction());
         for (ActionPlugin plugin : actionPlugins) {
             for (RestHandler handler : plugin.getRestHandlers(
                 settings,

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

@@ -0,0 +1,20 @@
+/*
+ * 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;
+
+public class CreatePITAction extends ActionType<CreatePITResponse> {
+    public static final CreatePITAction INSTANCE = new CreatePITAction();
+    public static final String NAME = "indices:data/write/pit";
+
+    private CreatePITAction() {
+        super(NAME, CreatePITResponse::new);
+    }
+}

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

@@ -0,0 +1,176 @@
+/*
+ * 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.ActionRequest;
+import org.opensearch.action.ActionRequestValidationException;
+import org.opensearch.action.IndicesRequest;
+import org.opensearch.action.support.IndicesOptions;
+import org.opensearch.common.Nullable;
+import org.opensearch.common.Strings;
+import org.opensearch.common.io.stream.StreamInput;
+import org.opensearch.common.io.stream.StreamOutput;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.tasks.Task;
+import org.opensearch.tasks.TaskId;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.opensearch.action.ValidateActions.addValidationError;
+
+/**
+ * A request to make create point in time against one or more indices.
+ */
+public class CreatePITRequest extends ActionRequest implements IndicesRequest.Replaceable {
+
+    // keep alive for pit reader context
+    private TimeValue keepAlive;
+
+    // this describes whether PIT can be created with partial failures
+    private Boolean allowPartialPitCreation;
+    @Nullable
+    private String routing = null;
+    @Nullable
+    private String preference = null;
+    private String[] indices = Strings.EMPTY_ARRAY;
+    private IndicesOptions indicesOptions = SearchRequest.DEFAULT_INDICES_OPTIONS;
+
+    public CreatePITRequest(TimeValue keepAlive, Boolean allowPartialPitCreation, String... indices) {
+        this.keepAlive = keepAlive;
+        this.allowPartialPitCreation = allowPartialPitCreation;
+        this.indices = indices;
+    }
+
+    public CreatePITRequest(StreamInput in) throws IOException {
+        super(in);
+        indices = in.readStringArray();
+        indicesOptions = IndicesOptions.readIndicesOptions(in);
+        routing = in.readOptionalString();
+        preference = in.readOptionalString();
+        keepAlive = in.readTimeValue();
+        routing = in.readOptionalString();
+        allowPartialPitCreation = in.readOptionalBoolean();
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeStringArray(indices);
+        indicesOptions.writeIndicesOptions(out);
+        out.writeOptionalString(preference);
+        out.writeTimeValue(keepAlive);
+        out.writeOptionalString(routing);
+        out.writeOptionalBoolean(allowPartialPitCreation);
+    }
+
+    public String getRouting() {
+        return routing;
+    }
+
+    public String getPreference() {
+        return preference;
+    }
+
+    public String[] getIndices() {
+        return indices;
+    }
+
+    public IndicesOptions getIndicesOptions() {
+        return indicesOptions;
+    }
+
+    public TimeValue getKeepAlive() {
+        return keepAlive;
+    }
+
+    /**
+     * Sets if this request should allow partial results.
+     */
+    public void allowPartialPitCreation(Boolean allowPartialPitCreation) {
+        this.allowPartialPitCreation = allowPartialPitCreation;
+    }
+
+    public boolean shouldAllowPartialPitCreation() {
+        return allowPartialPitCreation;
+    }
+
+    public void setRouting(String routing) {
+        this.routing = routing;
+    }
+
+    public void setPreference(String preference) {
+        this.preference = preference;
+    }
+
+    public void setIndices(String[] indices) {
+        this.indices = indices;
+    }
+
+    public void setIndicesOptions(IndicesOptions indicesOptions) {
+        this.indicesOptions = Objects.requireNonNull(indicesOptions, "indicesOptions must not be null");
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        ActionRequestValidationException validationException = null;
+        if (keepAlive == null) {
+            validationException = addValidationError("keep alive not specified", validationException);
+        }
+        return validationException;
+    }
+
+    @Override
+    public String[] indices() {
+        return indices;
+    }
+
+    @Override
+    public IndicesOptions indicesOptions() {
+        return indicesOptions;
+    }
+
+    public CreatePITRequest indicesOptions(IndicesOptions indicesOptions) {
+        this.indicesOptions = Objects.requireNonNull(indicesOptions, "indicesOptions must not be null");
+        return this;
+    }
+
+    public void setKeepAlive(TimeValue keepAlive) {
+        this.keepAlive = keepAlive;
+    }
+
+    public final String buildDescription() {
+        StringBuilder sb = new StringBuilder();
+        sb.append("indices[");
+        Strings.arrayToDelimitedString(indices, ",", sb);
+        sb.append("], ");
+        sb.append("pointintime[").append(keepAlive).append("], ");
+        return sb.toString();
+    }
+
+    @Override
+    public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+        return new Task(id, type, action, this.buildDescription(), parentTaskId, headers);
+    }
+
+    private void validateIndices(String... indices) {
+        Objects.requireNonNull(indices, "indices must not be null");
+        for (String index : indices) {
+            Objects.requireNonNull(index, "index must not be null");
+        }
+    }
+
+    @Override
+    public CreatePITRequest indices(String... indices) {
+        validateIndices(indices);
+        this.indices = indices;
+        return this;
+    }
+}