Merge branch 'main' into skip-can-match-on-batched

Author: original-brownbear

qa/smoke-test-http/src/internalClusterTest/java/org/elasticsearch/http/SearchRestCancellationIT.java

@@ -12,8 +12,6 @@
 import org.apache.http.nio.entity.NByteArrayEntity;
 import org.apache.logging.log4j.LogManager;
 import org.apache.lucene.util.SetOnce;
-import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
-import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
 import org.elasticsearch.action.admin.cluster.node.tasks.list.ListTasksResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.search.MultiSearchRequest;
@@ -46,7 +44,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CancellationException;
@@ -94,7 +91,7 @@ public void testAutomaticCancellationMultiSearchDuringQueryPhase() throws Except
     }
 
     void verifyCancellationDuringQueryPhase(String searchAction, Request searchRequest) throws Exception {
-        Map<String, String> nodeIdToName = readNodesInfo();
+        Map<String, String> nodeIdToName = nodeIdsToNames();
 
         List<ScriptedBlockPlugin> plugins = initBlockFactory();
         indexTestData();
@@ -137,7 +134,7 @@ public void testAutomaticCancellationMultiSearchDuringFetchPhase() throws Except
     }
 
     void verifyCancellationDuringFetchPhase(String searchAction, Request searchRequest) throws Exception {
-        Map<String, String> nodeIdToName = readNodesInfo();
+        Map<String, String> nodeIdToName = nodeIdsToNames();
 
         List<ScriptedBlockPlugin> plugins = initBlockFactory();
         indexTestData();
@@ -153,16 +150,6 @@ void verifyCancellationDuringFetchPhase(String searchAction, Request searchReque
         expectThrows(CancellationException.class, future::actionGet);
     }
 
-    private static Map<String, String> readNodesInfo() {
-        Map<String, String> nodeIdToName = new HashMap<>();
-        NodesInfoResponse nodesInfoResponse = clusterAdmin().prepareNodesInfo().get();
-        assertFalse(nodesInfoResponse.hasFailures());
-        for (NodeInfo node : nodesInfoResponse.getNodes()) {
-            nodeIdToName.put(node.getNode().getId(), node.getNode().getName());
-        }
-        return nodeIdToName;
-    }
-
     private static void ensureSearchTaskIsCancelled(String transportAction, Function<String, String> nodeIdToName) throws Exception {
         SetOnce<TaskInfo> searchTask = new SetOnce<>();
         ListTasksResponse listTasksResponse = clusterAdmin().prepareListTasks().get();

test/framework/src/main/java/org/elasticsearch/test/ESIntegTestCase.java

@@ -193,6 +193,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.IdentityHashMap;
 import java.util.List;
@@ -1207,6 +1208,19 @@ public static String getNodeId(String nodeName) {
         return internalCluster().getInstance(ClusterService.class, nodeName).localNode().getId();
     }
 
+    /**
+     * @return A map of the cluster node Ids to their node names.
+     */
+    public static Map<String, String> nodeIdsToNames() {
+        var names = internalCluster().getNodeNames();
+        Map<String, String> nodeIdsToNames = new HashMap<>();
+        for (var name : names) {
+            nodeIdsToNames.put(getNodeId(name), name);
+        }
+        return nodeIdsToNames;
+
+    }
+
     /**
      * Waits until at least a give number of document is visible for searchers
      *

test/framework/src/main/java/org/elasticsearch/test/InternalTestCluster.java

@@ -2314,6 +2314,11 @@ public int numMasterNodes() {
         return filterNodes(nodes, NodeAndClient::isMasterEligible).size();
     }
 
+    public Set<String> masterEligibleNodeNames() {
+        var masterEligibleNodes = filterNodes(nodes, NodeAndClient::isMasterEligible);
+        return masterEligibleNodes.stream().map(nodeAndClient -> nodeAndClient.name).collect(Collectors.toSet());
+    }
+
     public void setDisruptionScheme(ServiceDisruptionScheme scheme) {
         assert activeDisruptionScheme == null
             : "there is already and active disruption [" + activeDisruptionScheme + "]. call clearDisruptionScheme first";

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/common/MapPathExtractor.java

@@ -10,8 +10,10 @@
 import org.elasticsearch.common.Strings;
 
 import java.util.ArrayList;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.regex.Pattern;
 
 /**
@@ -78,6 +80,8 @@
  *   [1, 2]
  * ]
  * }
+ *
+ * The array field names would be {@code ["embeddings", "embedding"}
  * </pre>
  *
  * This implementation differs from JSONPath when handling a list of maps. JSONPath will flatten the result and return a single array.
@@ -123,10 +127,28 @@ public class MapPathExtractor {
     private static final String DOLLAR = "$";
 
     // default for testing
-    static final Pattern dotFieldPattern = Pattern.compile("^\\.([^.\\[]+)(.*)");
-    static final Pattern arrayWildcardPattern = Pattern.compile("^\\[\\*\\](.*)");
+    static final Pattern DOT_FIELD_PATTERN = Pattern.compile("^\\.([^.\\[]+)(.*)");
+    static final Pattern ARRAY_WILDCARD_PATTERN = Pattern.compile("^\\[\\*\\](.*)");
+    public static final String UNKNOWN_FIELD_NAME = "unknown";
+
+    /**
+     * A result object that tries to match up the field names parsed from the passed in path and the result
+     * extracted from the passed in map.
+     * @param extractedObject represents the extracted result from the map
+     * @param traversedFields a list of field names in order as they're encountered while navigating through the nested objects
+     */
+    public record Result(Object extractedObject, List<String> traversedFields) {
+        public String getArrayFieldName(int index) {
+            // if the index is out of bounds we'll return a default value
+            if (traversedFields.size() <= index || index < 0) {
+                return UNKNOWN_FIELD_NAME;
+            }
+
+            return traversedFields.get(index);
+        }
+    }
 
-    public static Object extract(Map<String, Object> data, String path) {
+    public static Result extract(Map<String, Object> data, String path) {
         if (data == null || data.isEmpty() || path == null || path.trim().isEmpty()) {
             return null;
         }
@@ -139,16 +161,41 @@ public static Object extract(Map<String, Object> data, String path) {
             throw new IllegalArgumentException(Strings.format("Path [%s] must start with a dollar sign ($)", cleanedPath));
         }
 
-        return navigate(data, cleanedPath);
+        var fieldNames = new LinkedHashSet<String>();
+
+        return new Result(navigate(data, cleanedPath, new FieldNameInfo("", "", fieldNames)), fieldNames.stream().toList());
     }
 
-    private static Object navigate(Object current, String remainingPath) {
-        if (current == null || remainingPath == null || remainingPath.isEmpty()) {
+    private record FieldNameInfo(String currentPath, String fieldName, Set<String> traversedFields) {
+        void addTraversedField(String fieldName) {
+            traversedFields.add(createPath(fieldName));
+        }
+
+        void addCurrentField() {
+            traversedFields.add(currentPath);
+        }
+
+        FieldNameInfo descend(String newFieldName) {
+            var newLocation = createPath(newFieldName);
+            return new FieldNameInfo(newLocation, newFieldName, traversedFields);
+        }
+
+        private String createPath(String newFieldName) {
+            if (Strings.isNullOrEmpty(currentPath)) {
+                return newFieldName;
+            } else {
+                return currentPath + "." + newFieldName;
+            }
+        }
+    }
+
+    private static Object navigate(Object current, String remainingPath, FieldNameInfo fieldNameInfo) {
+        if (current == null || Strings.isNullOrEmpty(remainingPath)) {
             return current;
         }
 
-        var dotFieldMatcher = dotFieldPattern.matcher(remainingPath);
-        var arrayWildcardMatcher = arrayWildcardPattern.matcher(remainingPath);
+        var dotFieldMatcher = DOT_FIELD_PATTERN.matcher(remainingPath);
+        var arrayWildcardMatcher = ARRAY_WILDCARD_PATTERN.matcher(remainingPath);
 
         if (dotFieldMatcher.matches()) {
             String field = dotFieldMatcher.group(1);
@@ -168,7 +215,12 @@ private static Object navigate(Object current, String remainingPath) {
                     throw new IllegalArgumentException(Strings.format("Unable to find field [%s] in map", field));
                 }
 
-                return navigate(currentMap.get(field), nextPath);
+                // Handle the case where the path was $.result.text or $.result[*].key
+                if (Strings.isNullOrEmpty(nextPath)) {
+                    fieldNameInfo.addTraversedField(field);
+                }
+
+                return navigate(currentMap.get(field), nextPath, fieldNameInfo.descend(field));
             } else {
                 throw new IllegalArgumentException(
                     Strings.format(
@@ -182,10 +234,12 @@ private static Object navigate(Object current, String remainingPath) {
         } else if (arrayWildcardMatcher.matches()) {
             String nextPath = arrayWildcardMatcher.group(1);
             if (current instanceof List<?> list) {
+                fieldNameInfo.addCurrentField();
+
                 List<Object> results = new ArrayList<>();
 
                 for (Object item : list) {
-                    Object result = navigate(item, nextPath);
+                    Object result = navigate(item, nextPath, fieldNameInfo);
                     if (result != null) {
                         results.add(result);
                     }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/external/http/retry/ErrorResponse.java

@@ -34,4 +34,16 @@ public String getErrorMessage() {
     public boolean errorStructureFound() {
         return errorStructureFound;
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        ErrorResponse that = (ErrorResponse) o;
+        return errorStructureFound == that.errorStructureFound && Objects.equals(errorMessage, that.errorMessage);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(errorMessage, errorStructureFound);
+    }
 }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/custom/CustomServiceSettings.java

@@ -0,0 +1,19 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.inference.services.custom;
+
+public class CustomServiceSettings {
+    public static final String NAME = "custom_service_settings";
+    public static final String URL = "url";
+    public static final String HEADERS = "headers";
+    public static final String REQUEST = "request";
+    public static final String REQUEST_CONTENT = "content";
+    public static final String RESPONSE = "response";
+    public static final String JSON_PARSER = "json_parser";
+    public static final String ERROR_PARSER = "error_parser";
+}