Validate PIT on _msearch (#63167)

This change ensures that we validate point in times provided by individual search
requests in _msearch.

Relates #63132

server/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java

@@ -28,6 +28,7 @@ import org.elasticsearch.common.CheckedBiConsumer;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.collect.Tuple;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContent;
 import org.elasticsearch.common.xcontent.XContentParser;
@@ -81,7 +82,7 @@ public class RestMultiSearchAction extends BaseRestHandler {
 
     @Override
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
-        final MultiSearchRequest multiSearchRequest = parseRequest(request, allowExplicitIndex);
+        final MultiSearchRequest multiSearchRequest = parseRequest(request, client.getNamedWriteableRegistry(), allowExplicitIndex);
         return channel -> {
             final RestCancellableNodeClient cancellableClient = new RestCancellableNodeClient(client, request.getHttpChannel());
             cancellableClient.execute(MultiSearchAction.INSTANCE, multiSearchRequest, new RestToXContentListener<>(channel));
@@ -91,7 +92,9 @@ public class RestMultiSearchAction extends BaseRestHandler {
     /**
      * Parses a {@link RestRequest} body and returns a {@link MultiSearchRequest}
      */
-    public static MultiSearchRequest parseRequest(RestRequest restRequest, boolean allowExplicitIndex) throws IOException {
+    public static MultiSearchRequest parseRequest(RestRequest restRequest,
+                                                  NamedWriteableRegistry namedWriteableRegistry,
+                                                  boolean allowExplicitIndex) throws IOException {
         MultiSearchRequest multiRequest = new MultiSearchRequest();
         IndicesOptions indicesOptions = IndicesOptions.fromRequest(restRequest, multiRequest.indicesOptions());
         multiRequest.indicesOptions(indicesOptions);
@@ -116,6 +119,13 @@ public class RestMultiSearchAction extends BaseRestHandler {
         parseMultiLineRequest(restRequest, multiRequest.indicesOptions(), allowExplicitIndex, (searchRequest, parser) -> {
             searchRequest.source(SearchSourceBuilder.fromXContent(parser, false));
             RestSearchAction.checkRestTotalHits(restRequest, searchRequest);
+            if (searchRequest.pointInTimeBuilder() != null) {
+                RestSearchAction.preparePointInTime(searchRequest, restRequest, namedWriteableRegistry);
+            } else {
+                searchRequest.setCcsMinimizeRoundtrips(
+                    restRequest.paramAsBoolean("ccs_minimize_roundtrips", searchRequest.isCcsMinimizeRoundtrips())
+                );
+            }
             multiRequest.add(searchRequest);
         });
         List<SearchRequest> requests = multiRequest.requests();

server/src/test/java/org/elasticsearch/action/search/MultiSearchRequestTests.java

@@ -95,7 +95,7 @@ public class MultiSearchRequestTests extends ESTestCase {
         FakeRestRequest restRequest = new FakeRestRequest.Builder(xContentRegistry())
             .withContent(new BytesArray(requestContent), XContentType.JSON).build();
         IllegalArgumentException ex = expectThrows(IllegalArgumentException.class,
-            () -> RestMultiSearchAction.parseRequest(restRequest, true));
+            () -> RestMultiSearchAction.parseRequest(restRequest, null, true));
         assertEquals("key [unknown_key] is not supported in the metadata section", ex.getMessage());
     }
 
@@ -104,7 +104,7 @@ public class MultiSearchRequestTests extends ESTestCase {
             "{\"query\" : {\"match_all\" :{}}}\r\n";
         FakeRestRequest restRequest = new FakeRestRequest.Builder(xContentRegistry())
             .withContent(new BytesArray(requestContent), XContentType.JSON).build();
-        MultiSearchRequest request = RestMultiSearchAction.parseRequest(restRequest, true);
+        MultiSearchRequest request = RestMultiSearchAction.parseRequest(restRequest, null, true);
         assertThat(request.requests().size(), equalTo(1));
         assertThat(request.requests().get(0).indices()[0], equalTo("test"));
         assertThat(request.requests().get(0).indicesOptions(),
@@ -118,7 +118,7 @@ public class MultiSearchRequestTests extends ESTestCase {
             .withContent(new BytesArray(requestContent), XContentType.JSON)
             .withParams(Collections.singletonMap("ignore_unavailable", "true"))
             .build();
-        MultiSearchRequest request = RestMultiSearchAction.parseRequest(restRequest, true);
+        MultiSearchRequest request = RestMultiSearchAction.parseRequest(restRequest, null, true);
         assertThat(request.requests().size(), equalTo(1));
         assertThat(request.requests().get(0).indices()[0], equalTo("test"));
         assertThat(request.requests().get(0).indicesOptions(),
@@ -209,13 +209,13 @@ public class MultiSearchRequestTests extends ESTestCase {
         RestRequest restRequest = new FakeRestRequest.Builder(xContentRegistry())
                 .withContent(new BytesArray(mserchAction.getBytes(StandardCharsets.UTF_8)), XContentType.JSON).build();
         IllegalArgumentException expectThrows = expectThrows(IllegalArgumentException.class,
-                () -> RestMultiSearchAction.parseRequest(restRequest, true));
+                () -> RestMultiSearchAction.parseRequest(restRequest, null, true));
         assertEquals("The msearch request must be terminated by a newline [\n]", expectThrows.getMessage());
 
         String mserchActionWithNewLine = mserchAction + "\n";
         RestRequest restRequestWithNewLine = new FakeRestRequest.Builder(xContentRegistry())
                 .withContent(new BytesArray(mserchActionWithNewLine.getBytes(StandardCharsets.UTF_8)), XContentType.JSON).build();
-        MultiSearchRequest msearchRequest = RestMultiSearchAction.parseRequest(restRequestWithNewLine, true);
+        MultiSearchRequest msearchRequest = RestMultiSearchAction.parseRequest(restRequestWithNewLine, null, true);
         assertEquals(3, msearchRequest.requests().size());
     }
 

x-pack/plugin/src/test/resources/rest-api-spec/test/data_stream/10_data_stream_resolvability.yml

@@ -648,7 +648,6 @@
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -659,7 +658,7 @@
             id: "$point_in_time_id"
             keep_alive: 1m
 
-  - match: {hits.total: 3 }
+  - match: {hits.total.value: 3 }
   - length: {hits.hits: 1 }
   - match: {hits.hits.0._index: .ds-simple-data-stream1-000001 }
   - match: {hits.hits.0._id: "123" }
@@ -667,7 +666,6 @@
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -678,7 +676,7 @@
           pit:
             id: "$point_in_time_id"
 
-  - match: {hits.total: 3}
+  - match: {hits.total.value: 3}
   - length: {hits.hits: 1 }
   - match: {hits.hits.0._index: .ds-simple-data-stream1-000001 }
   - match: {hits.hits.0._id: "5" }
@@ -686,7 +684,6 @@
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -698,7 +695,7 @@
             id: "$point_in_time_id"
             keep_alive: 1m
 
-  - match: {hits.total: 3}
+  - match: {hits.total.value: 3}
   - length: {hits.hits: 1 }
   - match: {hits.hits.0._index: .ds-simple-data-stream1-000001 }
   - match: {hits.hits.0._id: "1" }
@@ -706,7 +703,6 @@
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -718,7 +714,7 @@
             id: "$point_in_time_id"
             keep_alive: 1m
 
-  - match: {hits.total: 3}
+  - match: {hits.total.value: 3}
   - length: {hits.hits: 0 }
 
   - do:

x-pack/plugin/src/test/resources/rest-api-spec/test/search/point_in_time.yml

@@ -47,7 +47,6 @@ setup:
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -58,7 +57,7 @@ setup:
             id: "$point_in_time_id"
             keep_alive: 1m
 
-  - match: {hits.total: 3 }
+  - match: {hits.total.value: 3 }
   - length: {hits.hits: 1 }
   - match: {hits.hits.0._index: test }
   - match: {hits.hits.0._id: "172" }
@@ -76,7 +75,6 @@ setup:
   # search with a point in time
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -87,7 +85,7 @@ setup:
           pit:
             id: "$point_in_time_id"
 
-  - match: {hits.total: 3 }
+  - match: {hits.total.value: 3 }
   - length: {hits.hits: 1 }
   - match: {hits.hits.0._index: test }
   - match: {hits.hits.0._id: "42" }
@@ -95,7 +93,6 @@ setup:
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -106,7 +103,7 @@ setup:
           pit:
             id: "$point_in_time_id"
 
-  - match: {hits.total: 3}
+  - match: {hits.total.value: 3}
   - length: {hits.hits: 1 }
   - match: {hits.hits.0._index: test }
   - match: {hits.hits.0._id: "1" }
@@ -114,7 +111,6 @@ setup:
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 1
           query:
@@ -126,7 +122,7 @@ setup:
             id: "$point_in_time_id"
             keep_alive: 1m
 
-  - match: {hits.total: 3}
+  - match: {hits.total.value: 3}
   - length: {hits.hits: 0 }
 
   - do:
@@ -147,7 +143,6 @@ setup:
 
   - do:
       search:
-        rest_total_hits_as_int: true
         body:
           size: 2
           query:
@@ -158,7 +153,7 @@ setup:
             id: "$point_in_time_id"
             keep_alive: 1m
 
-  - match: {hits.total: 4 }
+  - match: {hits.total.value: 4 }
   - length: {hits.hits: 2 }
   - match: {hits.hits.0._index: test }
   - match: {hits.hits.0._id: "172" }
@@ -169,3 +164,41 @@ setup:
       close_point_in_time:
         body:
           id: "$point_in_time_id"
+
+---
+"msearch":
+  - skip:
+      version: " - 7.99.99"
+      reason: "After backport: 7.9.99 => point in time is introduced in 7.10"
+  - do:
+      open_point_in_time:
+        index: "t*"
+        keep_alive: 5m
+  - set: {id: point_in_time_id}
+
+  - do:
+      msearch:
+        body:
+          - {}
+          - { query: { match: { _index: test }}, pit: { id: "$point_in_time_id" }}
+          - {}
+          - { query: { match_all: {}}, pit: { id: "$point_in_time_id" }}
+
+  - match:  { responses.0.hits.total.value:     3   }
+  - match:  { responses.0.hits.total.relation:  eq  }
+  - match:  { responses.1.hits.total.value:     4   }
+  - match:  { responses.1.hits.total.relation:  eq  }
+
+  - do:
+      catch: bad_request
+      msearch:
+        body:
+          - index: index
+          - { query: { match: { foo: bar }}, pit: { id: "$point_in_time_id" }}
+          - index: index2
+          - { query: { match_all: {}}, pit: { id: "$point_in_time_id" }}
+
+  - do:
+      close_point_in_time:
+        body:
+          id: "$point_in_time_id"