Merge branch 'main' into esql/project_routing

Author: luigidellaquila

docs/changelog/133675.yaml

@@ -0,0 +1,5 @@
+pr: 133675
+summary: Support using the semantic query across multiple inference IDs
+area: Vector Search
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -356,6 +356,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion PROJECT_RESERVED_STATE_MOVE_TO_REGISTRY = def(9_147_0_00);
     public static final TransportVersion STREAMS_ENDPOINT_PARAM_RESTRICTIONS = def(9_148_0_00);
     public static final TransportVersion RESOLVE_INDEX_MODE_FILTER = def(9_149_0_00);
+    public static final TransportVersion SEMANTIC_QUERY_MULTIPLE_INFERENCE_IDS = def(9_150_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/SecurityExtension.java

@@ -16,7 +16,7 @@
 import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationFailureHandler;
 import org.elasticsearch.xpack.core.security.authc.Realm;
-import org.elasticsearch.xpack.core.security.authc.apikey.CustomApiKeyAuthenticator;
+import org.elasticsearch.xpack.core.security.authc.apikey.CustomAuthenticator;
 import org.elasticsearch.xpack.core.security.authc.service.NodeLocalServiceAccountTokenStore;
 import org.elasticsearch.xpack.core.security.authc.service.ServiceAccountTokenStore;
 import org.elasticsearch.xpack.core.security.authc.support.UserRoleMapper;
@@ -129,7 +129,7 @@ default ServiceAccountTokenStore getServiceAccountTokenStore(SecurityComponents
         return null;
     }
 
-    default CustomApiKeyAuthenticator getCustomApiKeyAuthenticator(SecurityComponents components) {
+    default List<CustomAuthenticator> getCustomAuthenticators(SecurityComponents components) {
         return null;
     }
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/Authentication.java

@@ -1376,6 +1376,18 @@ public static Authentication newRealmAuthentication(User user, RealmRef realmRef
         return authentication;
     }
 
+    public static Authentication newCloudAccessTokenAuthentication(
+        AuthenticationResult<User> authResult,
+        Authentication.RealmRef realmRef
+    ) {
+        assert authResult.isAuthenticated() : "cloud access token authn result must be successful";
+        final User user = authResult.getValue();
+        return new Authentication(
+            new Subject(user, realmRef, TransportVersion.current(), authResult.getMetadata()),
+            AuthenticationType.TOKEN
+        );
+    }
+
     public static Authentication newCloudApiKeyAuthentication(AuthenticationResult<User> authResult, String nodeName) {
         assert authResult.isAuthenticated() : "cloud API Key authn result must be successful";
         final User apiKeyUser = authResult.getValue();

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/apikey/CustomApiKeyAuthenticator.java

@@ -0,0 +1,31 @@
+/*
+ * 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.core.security.authc.apikey;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.core.Nullable;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationResult;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationToken;
+
+/**
+ * An extension point to provide a custom authenticator implementation. For example, a custom API key or a custom OAuth2
+ * token implementation. The implementation is wrapped by a core `Authenticator` class and included in the authenticator chain
+ * _before_ the respective "standard" authenticator(s).
+ */
+public interface CustomAuthenticator {
+
+    boolean supports(AuthenticationToken token);
+
+    @Nullable
+    AuthenticationToken extractToken(ThreadContext context);
+
+    void authenticate(@Nullable AuthenticationToken token, ActionListener<AuthenticationResult<Authentication>> listener);
+
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/apikey/CustomAuthenticator.java

@@ -34,15 +34,28 @@ public abstract class SemanticMatchTestCase extends ESRestTestCase {
     public void testWithMultipleInferenceIds() throws IOException {
         assumeTrue("semantic text capability not available", EsqlCapabilities.Cap.SEMANTIC_TEXT_FIELD_CAPS.isEnabled());
 
+        var request1 = new Request("POST", "/test-semantic1/_doc/id-1");
+        request1.addParameter("refresh", "true");
+        request1.setJsonEntity("{\"semantic_text_field\": \"inference test 1\"}");
+        assertEquals(201, adminClient().performRequest(request1).getStatusLine().getStatusCode());
+
+        var request2 = new Request("POST", "/test-semantic2/_doc/id-2");
+        request2.addParameter("refresh", "true");
+        request2.setJsonEntity("{\"semantic_text_field\": \"inference test 2\"}");
+        assertEquals(201, adminClient().performRequest(request2).getStatusLine().getStatusCode());
+
         String query = """
             from test-semantic1,test-semantic2
             | where match(semantic_text_field, "something")
+            | SORT semantic_text_field ASC
             """;
-        ResponseException re = expectThrows(ResponseException.class, () -> runEsqlQuery(query));
-
-        assertThat(re.getMessage(), containsString("Field [semantic_text_field] has multiple inference IDs associated with it"));
+        Map<String, Object> result = runEsqlQuery(query);
 
-        assertEquals(400, re.getResponse().getStatusLine().getStatusCode());
+        assertResultMap(
+            result,
+            matchesList().item(matchesMap().entry("name", "semantic_text_field").entry("type", "text")),
+            matchesList(List.of(List.of("inference test 1"), List.of("inference test 2")))
+        );
     }
 
     public void testWithInferenceNotConfigured() {
@@ -128,6 +141,28 @@ public void setUpIndices() throws IOException {
         createIndex(adminClient(), "test-semantic4", settings, mapping4);
     }
 
+    @Before
+    public void setUpSparseEmbeddingInferenceEndpoint() throws IOException {
+        Request request = new Request("PUT", "_inference/sparse_embedding/test_sparse_inference");
+        request.setJsonEntity("""
+                  {
+                   "service": "test_service",
+                   "service_settings": {
+                     "model": "my_model",
+                     "api_key": "abc64"
+                   },
+                   "task_settings": {
+                   }
+                 }
+            """);
+        try {
+            adminClient().performRequest(request);
+        } catch (ResponseException exc) {
+            // in case the removal failed
+            assertThat(exc.getResponse().getStatusLine().getStatusCode(), equalTo(400));
+        }
+    }
+
     @Before
     public void setUpTextEmbeddingInferenceEndpoint() throws IOException {
         Request request = new Request("PUT", "_inference/text_embedding/test_dense_inference");
@@ -155,6 +190,15 @@ public void setUpTextEmbeddingInferenceEndpoint() throws IOException {
     public void wipeData() throws IOException {
         adminClient().performRequest(new Request("DELETE", "*"));
 
+        try {
+            adminClient().performRequest(new Request("DELETE", "_inference/test_sparse_inference"));
+        } catch (ResponseException e) {
+            // 404 here means the endpoint was not created
+            if (e.getResponse().getStatusLine().getStatusCode() != 404) {
+                throw e;
+            }
+        }
+
         try {
             adminClient().performRequest(new Request("DELETE", "_inference/test_dense_inference"));
         } catch (ResponseException e) {

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/SemanticMatchTestCase.java

@@ -96,6 +96,10 @@ public First withFilter(Expression filter) {
         return new First(source(), field(), filter, sort);
     }
 
+    public Expression sort() {
+        return sort;
+    }
+
     @Override
     public DataType dataType() {
         return field().dataType().noText();

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/First.java

@@ -96,6 +96,10 @@ public Last withFilter(Expression filter) {
         return new Last(source(), field(), filter, sort);
     }
 
+    public Expression sort() {
+        return sort;
+    }
+
     @Override
     public DataType dataType() {
         return field().dataType().noText();

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/Last.java

@@ -506,41 +506,46 @@ private void processPageGrouping(GroupingAggregator aggregator, Page inputPage,
     }
 
     private void assertAggregatorToString(Object aggregator) {
-        if (optIntoToAggregatorToStringChecks() == false) {
-            return;
-        }
         String expectedStart = switch (aggregator) {
             case Aggregator a -> "Aggregator[aggregatorFunction=";
             case GroupingAggregator a -> "GroupingAggregator[aggregatorFunction=";
             default -> throw new UnsupportedOperationException("can't check toString for [" + aggregator.getClass() + "]");
         };
+        String channels = initialInputChannels().stream().map(Object::toString).collect(Collectors.joining(", "));
         String expectedEnd = switch (aggregator) {
-            case Aggregator a -> "AggregatorFunction[channels=[0]], mode=SINGLE]";
-            case GroupingAggregator a -> "GroupingAggregatorFunction[channels=[0]], mode=SINGLE]";
+            case Aggregator a -> "AggregatorFunction[channels=[" + channels + "]], mode=SINGLE]";
+            case GroupingAggregator a -> "GroupingAggregatorFunction[channels=[" + channels + "]], mode=SINGLE]";
             default -> throw new UnsupportedOperationException("can't check toString for [" + aggregator.getClass() + "]");
         };
 
         String toString = aggregator.toString();
         assertThat(toString, startsWith(expectedStart));
-        assertThat(toString.substring(expectedStart.length(), toString.length() - expectedEnd.length()), testCase.evaluatorToString());
         assertThat(toString, endsWith(expectedEnd));
-    }
-
-    protected boolean optIntoToAggregatorToStringChecks() {
-        // TODO remove this when everyone has opted in
-        return false;
+        assertThat(toString.substring(expectedStart.length(), toString.length() - expectedEnd.length()), testCase.evaluatorToString());
     }
 
     protected static String standardAggregatorName(String prefix, DataType type) {
         String typeName = switch (type) {
             case BOOLEAN -> "Boolean";
+            case CARTESIAN_POINT -> "CartesianPoint";
+            case CARTESIAN_SHAPE -> "CartesianShape";
+            case GEO_POINT -> "GeoPoint";
+            case GEO_SHAPE -> "GeoShape";
             case KEYWORD, TEXT, VERSION -> "BytesRef";
-            case DOUBLE -> "Double";
-            case INTEGER -> "Int";
+            case DOUBLE, COUNTER_DOUBLE -> "Double";
+            case INTEGER, COUNTER_INTEGER -> "Int";
             case IP -> "Ip";
-            case DATETIME, DATE_NANOS, LONG, UNSIGNED_LONG -> "Long";
+            case DATETIME, DATE_NANOS, LONG, COUNTER_LONG, UNSIGNED_LONG -> "Long";
+            case NULL -> "Null";
             default -> throw new UnsupportedOperationException("name for [" + type + "]");
         };
         return prefix + typeName;
     }
+
+    protected static String standardAggregatorNameAllBytesTheSame(String prefix, DataType type) {
+        return standardAggregatorName(prefix, switch (type) {
+            case CARTESIAN_POINT, CARTESIAN_SHAPE, GEO_POINT, GEO_SHAPE, IP -> DataType.KEYWORD;
+            default -> type;
+        });
+    }
 }