remove ignoreFailure and fix JsonArray Parsing Issue (#2770) (#2774)

Author: opensearch-trigger-bot[bot]

plugin/src/main/java/org/opensearch/ml/processor/MLInferenceSearchRequestProcessor.java

@@ -251,7 +251,7 @@ public void onResponse(Map<Integer, MLOutput> multipleMLOutputs) {
                             requestListener.onResponse(request);
                         }
                     } catch (Exception e) {
-                        if (ignoreMissing || ignoreFailure) {
+                        if (ignoreFailure) {
                             logger.error("Failed in writing prediction outcomes to new query", e);
                             requestListener.onResponse(request);
 
@@ -348,7 +348,7 @@ private boolean validateQueryFieldInQueryString(
             for (Map.Entry<String, String> entry : inputMap.entrySet()) {
                 // the inputMap takes in model input as keys and query fields as value
                 String queryField = entry.getValue();
-                String pathData = jsonData.read(queryField);
+                Object pathData = jsonData.read(queryField);
                 if (pathData == null) {
                     throw new IllegalArgumentException("cannot find field: " + queryField + " in query string: " + jsonData.jsonString());
                 }
@@ -358,7 +358,7 @@ private boolean validateQueryFieldInQueryString(
             for (Map<String, String> outputMap : processOutputMap) {
                 for (Map.Entry<String, String> entry : outputMap.entrySet()) {
                     String queryField = entry.getKey();
-                    String pathData = jsonData.read(queryField);
+                    Object pathData = jsonData.read(queryField);
                     if (pathData == null) {
                         throw new IllegalArgumentException(
                             "cannot find field: " + queryField + " in query string: " + jsonData.jsonString()
@@ -402,7 +402,7 @@ private void processPredictions(
                 // model field as key, query field name as value
                 String modelInputFieldName = entry.getKey();
                 String queryFieldName = entry.getValue();
-                String queryFieldValue = JsonPath.parse(newQuery).read(queryFieldName);
+                String queryFieldValue = StringUtils.toJson(JsonPath.parse(newQuery).read(queryFieldName));
                 modelParameters.put(modelInputFieldName, queryFieldValue);
             }
         }

plugin/src/main/java/org/opensearch/ml/processor/MLInferenceSearchResponseProcessor.java

@@ -478,7 +478,6 @@ public void onFailure(Exception e) {
      */
     private boolean checkIsModelInputMissing(Map<String, Object> document, Map<String, String> inputMapping) {
         boolean isModelInputMissing = false;
-
         for (Map.Entry<String, String> inputMapEntry : inputMapping.entrySet()) {
             String oldDocumentFieldName = inputMapEntry.getValue();
             boolean checkSingleModelInputPresent = hasField(document, oldDocumentFieldName);

plugin/src/test/java/org/opensearch/ml/processor/MLInferenceSearchRequestProcessorTests.java

@@ -28,6 +28,7 @@
 import org.opensearch.index.query.QueryBuilder;
 import org.opensearch.index.query.RangeQueryBuilder;
 import org.opensearch.index.query.TermQueryBuilder;
+import org.opensearch.index.query.TermsQueryBuilder;
 import org.opensearch.ingest.Processor;
 import org.opensearch.ml.common.output.model.ModelTensor;
 import org.opensearch.ml.common.output.model.ModelTensorOutput;
@@ -247,6 +248,66 @@ public void onFailure(Exception e) {
 
     }
 
+    /**
+     * Tests the successful rewriting of multiple string in terms query based on the model output.
+     *
+     * @throws Exception if an error occurs during the test
+     */
+    public void testExecute_rewriteTermsQuerySuccess() throws Exception {
+        /**
+         * example term query: {"query":{"terms":{"text":["foo","bar],"boost":1.0}}}
+         */
+        String modelInputField = "inputs";
+        String originalQueryField = "query.terms.text";
+        String newQueryField = "query.terms.text";
+        String modelOutputField = "response";
+        MLInferenceSearchRequestProcessor requestProcessor = getMlInferenceSearchRequestProcessor(
+            null,
+            modelInputField,
+            originalQueryField,
+            newQueryField,
+            modelOutputField,
+            false,
+            false
+        );
+        ModelTensor modelTensor = ModelTensor
+            .builder()
+            .dataAsMap(ImmutableMap.of("response", Arrays.asList("car", "vehicle", "truck")))
+            .build();
+        ModelTensors modelTensors = ModelTensors.builder().mlModelTensors(Arrays.asList(modelTensor)).build();
+        ModelTensorOutput mlModelTensorOutput = ModelTensorOutput.builder().mlModelOutputs(Arrays.asList(modelTensors)).build();
+
+        doAnswer(invocation -> {
+            ActionListener<MLTaskResponse> actionListener = invocation.getArgument(2);
+            actionListener.onResponse(MLTaskResponse.builder().output(mlModelTensorOutput).build());
+            return null;
+        }).when(client).execute(any(), any(), any());
+
+        QueryBuilder incomingQuery = new TermsQueryBuilder("text", Arrays.asList("foo", "bar"));
+        SearchSourceBuilder source = new SearchSourceBuilder().query(incomingQuery);
+        SearchRequest request = new SearchRequest().source(source);
+        /**
+         * example terms query: {"query":{"terms":{"text":["car","vehicle","truck"],"boost":1.0}}}
+         */
+
+        ActionListener<SearchRequest> Listener = new ActionListener<>() {
+            @Override
+            public void onResponse(SearchRequest newSearchRequest) {
+                QueryBuilder expectedQuery = new TermsQueryBuilder("text", Arrays.asList("car", "vehicle", "truck"));
+                assertEquals(expectedQuery, newSearchRequest.source().query());
+                assertEquals(request.toString(), newSearchRequest.toString());
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                throw new RuntimeException("Failed in executing processRequestAsync.");
+            }
+        };
+
+        requestProcessor.processRequestAsync(request, requestContext, Listener);
+
+    }
+
     /**
      * Tests the successful rewriting of a double in a term query based on the model output.
      *
@@ -444,7 +505,6 @@ public void onFailure(Exception e) {
      * @throws Exception if an error occurs during the test
      */
     public void testExecute_rewriteListFromTermQueryToGeometryQuerySuccess() throws Exception {
-
         String queryTemplate = "{\n"
             + "  \"query\": {\n"
             + "  \"geo_shape\" : {\n"