Create PIT API

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

@@ -235,14 +235,7 @@
 import org.opensearch.action.ingest.SimulatePipelineTransportAction;
 import org.opensearch.action.main.MainAction;
 import org.opensearch.action.main.TransportMainAction;
-import org.opensearch.action.search.ClearScrollAction;
-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.TransportMultiSearchAction;
-import org.opensearch.action.search.TransportSearchAction;
-import org.opensearch.action.search.TransportSearchScrollAction;
+import org.opensearch.action.search.*;
 import org.opensearch.action.support.ActionFilters;
 import org.opensearch.action.support.AutoCreateIndex;
 import org.opensearch.action.support.DestructiveOperations;
@@ -398,12 +391,7 @@
 import org.opensearch.rest.action.ingest.RestGetPipelineAction;
 import org.opensearch.rest.action.ingest.RestPutPipelineAction;
 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.RestExplainAction;
-import org.opensearch.rest.action.search.RestMultiSearchAction;
-import org.opensearch.rest.action.search.RestSearchAction;
-import org.opensearch.rest.action.search.RestSearchScrollAction;
+import org.opensearch.rest.action.search.*;
 import org.opensearch.tasks.Task;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.usage.UsageService;
@@ -660,6 +648,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 +821,10 @@ 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<SearchResponse> {
+    public static final CreatePITAction INSTANCE = new CreatePITAction();
+    public static final String NAME = "indices:data/read/pit";
+
+    private CreatePITAction() {
+        super(NAME, SearchResponse::new);
+    }
+}

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

@@ -0,0 +1,139 @@
+/*
+ * 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;
+
+public class CreatePITRequest extends ActionRequest implements IndicesRequest.Replaceable {
+
+    private TimeValue keepAlive;
+    private final 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) {
+        this.keepAlive = keepAlive;
+        this.allowPartialPitCreation = allowPartialPitCreation;
+    }
+
+    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.readBoolean();
+    }
+
+    @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.writeBoolean(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;
+    }
+
+    public boolean isAllowPartialPitCreation() {
+        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() {
+        return null;
+    }
+
+    @Override
+    public String[] indices() {
+        return indices;
+    }
+
+    @Override
+    public IndicesOptions indicesOptions() {
+        return indicesOptions;
+    }
+
+    public void setKeepAlive(TimeValue keepAlive) {
+        this.keepAlive = keepAlive;
+    }
+
+    @Override
+    public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
+        return new SearchTask(id, type, action, () -> "desc", parentTaskId, headers);
+    }
+
+    /**
+     * Sets the indices the search will be executed on.
+     */
+    @Override
+    public CreatePITRequest indices(String... indices) {
+        SearchRequest.validateIndices(indices);
+        this.indices = indices;
+        return this;
+    }
+}

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

@@ -379,7 +379,7 @@ public SearchRequest indices(String... indices) {
         return this;
     }
 
-    private static void validateIndices(String... indices) {
+    protected static void validateIndices(String... indices) {
         Objects.requireNonNull(indices, "indices must not be null");
         for (String index : indices) {
             Objects.requireNonNull(index, "index must not be null");

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

@@ -93,6 +93,8 @@ public class SearchTransportService {
     public static final String FETCH_ID_SCROLL_ACTION_NAME = "indices:data/read/search[phase/fetch/id/scroll]";
     public static final String FETCH_ID_ACTION_NAME = "indices:data/read/search[phase/fetch/id]";
     public static final String QUERY_CAN_MATCH_NAME = "indices:data/read/search[can_match]";
+    public static final String CREATE_READER_CONTEXT_ACTION_NAME = "indices:data/read/search[create_context]";
+    public static final String UPDATE_READER_CONTEXT_ACTION_NAME = "indices:data/read/search[update_context]";
 
     private final TransportService transportService;
     private final BiFunction<Transport.Connection, SearchActionListener, ActionListener> responseWrapper;
@@ -140,6 +142,20 @@ public void sendFreeContext(
         );
     }
 
+    public void updatePitContext(
+            Transport.Connection connection,
+            TransportCreatePITAction.UpdatePITReaderRequest request,
+            ActionListener<TransportCreatePITAction.UpdatePitContextResponse> actionListener) {
+        transportService.sendRequest(
+                connection,
+                UPDATE_READER_CONTEXT_ACTION_NAME,
+                request,
+                TransportRequestOptions.EMPTY,
+                new ActionListenerResponseHandler<TransportCreatePITAction.UpdatePitContextResponse>(actionListener,
+                        TransportCreatePITAction.UpdatePitContextResponse::new)
+        );
+    }
+
     public void sendCanMatch(
         Transport.Connection connection,
         final ShardSearchRequest request,
@@ -307,7 +323,7 @@ public Map<String, Long> getPendingSearchRequests() {
         return new HashMap<>(clientConnections);
     }
 
-    static class ScrollFreeContextRequest extends TransportRequest {
+  static class ScrollFreeContextRequest extends TransportRequest {
         private ShardSearchContextId contextId;
 
         ScrollFreeContextRequest(ShardSearchContextId contextId) {
@@ -545,6 +561,50 @@ public static void registerRequestHandler(TransportService transportService, Sea
             }
         );
         TransportActionProxy.registerProxyAction(transportService, QUERY_CAN_MATCH_NAME, SearchService.CanMatchResponse::new);
+        transportService.registerRequestHandler(
+            CREATE_READER_CONTEXT_ACTION_NAME,
+            ThreadPool.Names.SAME,
+            TransportCreatePITAction.CreateReaderContextRequest::new,
+            (request, channel, task) -> {
+              ChannelActionListener<
+                      TransportCreatePITAction.CreateReaderContextResponse,
+                      TransportCreatePITAction.CreateReaderContextRequest>
+                  listener =
+                      new ChannelActionListener<>(channel, CREATE_READER_CONTEXT_ACTION_NAME, request);
+              searchService.openReaderContext(
+                  request.getShardId(),
+                  request.getKeepAlive(),
+                  ActionListener.wrap(
+                      r ->
+                          listener.onResponse(
+                              new TransportCreatePITAction.CreateReaderContextResponse(r)),
+                      listener::onFailure));
+            });
+        TransportActionProxy.registerProxyAction(
+            transportService,
+            CREATE_READER_CONTEXT_ACTION_NAME,
+            TransportCreatePITAction.CreateReaderContextResponse::new);
+
+        transportService.registerRequestHandler(
+            UPDATE_READER_CONTEXT_ACTION_NAME,
+            ThreadPool.Names.SAME,
+            TransportCreatePITAction.UpdatePITReaderRequest::new,
+            (request, channel, task) -> {
+                ChannelActionListener<TransportCreatePITAction.UpdatePitContextResponse,
+                        TransportCreatePITAction.UpdatePITReaderRequest> listener =
+                    new ChannelActionListener<>(channel, UPDATE_READER_CONTEXT_ACTION_NAME, request);
+                searchService.updatePitIdAndKeepAlive(request,
+                    ActionListener.wrap(
+                        r -> listener.onResponse(r),
+                        listener::onFailure
+                    ));
+            }
+        );
+        TransportActionProxy.registerProxyAction(
+                transportService,
+                UPDATE_READER_CONTEXT_ACTION_NAME,
+                TransportCreatePITAction.UpdatePitContextResponse::new);
+
     }
 
     /**