Merge branch 'main' into handle-9/deprecate-node-attrs-v2-follow-up

Author: gmarouli

docs/changelog/121260.yaml

@@ -0,0 +1,5 @@
+pr: 121260
+summary: Introduce a pre-mapping logical plan processing step
+area: ES|QL
+type: enhancement
+issues: []

muted-tests.yml

@@ -231,9 +231,6 @@ tests:
 - class: org.elasticsearch.datastreams.DataStreamsClientYamlTestSuiteIT
   method: test {p0=data_stream/140_data_stream_aliases/Create data stream aliases using wildcard expression}
   issue: https://github.com/elastic/elasticsearch/issues/120890
-- class: org.elasticsearch.xpack.test.rest.XPackRestIT
-  method: test {p0=ml/inference_crud/*}
-  issue: https://github.com/elastic/elasticsearch/issues/120816
 - class: org.elasticsearch.xpack.security.authc.service.ServiceAccountIT
   method: testAuthenticateShouldNotFallThroughInCaseOfFailure
   issue: https://github.com/elastic/elasticsearch/issues/120902
@@ -330,6 +327,10 @@ tests:
   method: testCrossClusterAsyncQueryStop
   issue: https://github.com/elastic/elasticsearch/issues/121249
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
+  method: test {p0=transform/*}
+  issue: https://github.com/elastic/elasticsearch/issues/120816
+- class: org.elasticsearch.xpack.test.rest.XPackRestIT
+  method: test {p0=ml/*}
   issue: https://github.com/elastic/elasticsearch/issues/120816
 - class: org.elasticsearch.upgrades.VectorSearchIT
   method: testBBQVectorSearch {upgradedNodes=0}

server/src/internalClusterTest/java/org/elasticsearch/action/admin/indices/mapping/put/PutMappingIT.java

@@ -0,0 +1,55 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.admin.indices.mapping.put;
+
+import org.apache.logging.log4j.Level;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.test.ESSingleNodeTestCase;
+import org.elasticsearch.test.MockLog;
+import org.elasticsearch.test.junit.annotations.TestLogging;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class PutMappingIT extends ESSingleNodeTestCase {
+
+    @TestLogging(
+        reason = "testing DEBUG logging",
+        value = "org.elasticsearch.action.admin.indices.mapping.put.TransportPutMappingAction:DEBUG"
+    )
+    public void testFailureLogging() {
+        final var indexName = randomIdentifier();
+        createIndex(indexName);
+        final var fieldName = randomIdentifier();
+        safeGet(client().execute(TransportPutMappingAction.TYPE, new PutMappingRequest(indexName).source(fieldName, "type=keyword")));
+        MockLog.assertThatLogger(
+            () -> assertThat(
+                asInstanceOf(
+                    IllegalArgumentException.class,
+                    safeAwaitFailure(
+                        AcknowledgedResponse.class,
+                        l -> client().execute(
+                            TransportPutMappingAction.TYPE,
+                            new PutMappingRequest(indexName).source(fieldName, "type=long"),
+                            l
+                        )
+                    )
+                ).getMessage(),
+                equalTo("mapper [" + fieldName + "] cannot be changed from type [keyword] to [long]")
+            ),
+            TransportPutMappingAction.class,
+            new MockLog.SeenEventExpectation(
+                "failure message",
+                TransportPutMappingAction.class.getCanonicalName(),
+                Level.DEBUG,
+                "failed to put mappings on indices [[" + indexName
+            )
+        );
+    }
+}

server/src/main/java/org/elasticsearch/action/admin/indices/mapping/put/TransportPutMappingAction.java

@@ -36,7 +36,6 @@
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -126,7 +125,7 @@ protected void masterOperation(
 
             performMappingUpdate(concreteIndices, request, listener, metadataMappingService, false);
         } catch (IndexNotFoundException ex) {
-            logger.debug(() -> "failed to put mappings on indices [" + Arrays.asList(request.indices() + "]"), ex);
+            logger.debug(() -> "failed to put mappings on indices " + Arrays.toString(request.indices()), ex);
             throw ex;
         }
     }
@@ -162,25 +161,21 @@ static void performMappingUpdate(
         MetadataMappingService metadataMappingService,
         boolean autoUpdate
     ) {
-        final ActionListener<AcknowledgedResponse> wrappedListener = listener.delegateResponse((l, e) -> {
-            logger.debug(() -> "failed to put mappings on indices [" + Arrays.asList(concreteIndices) + "]", e);
+        ActionListener.run(listener.delegateResponse((l, e) -> {
+            logger.debug(() -> "failed to put mappings on indices " + Arrays.toString(concreteIndices), e);
             l.onFailure(e);
-        });
-        final PutMappingClusterStateUpdateRequest updateRequest;
-        try {
-            updateRequest = new PutMappingClusterStateUpdateRequest(
-                request.masterNodeTimeout(),
-                request.ackTimeout(),
-                request.source(),
-                autoUpdate,
-                concreteIndices
-            );
-        } catch (IOException e) {
-            wrappedListener.onFailure(e);
-            return;
-        }
-
-        metadataMappingService.putMapping(updateRequest, wrappedListener);
+        }),
+            wrappedListener -> metadataMappingService.putMapping(
+                new PutMappingClusterStateUpdateRequest(
+                    request.masterNodeTimeout(),
+                    request.ackTimeout(),
+                    request.source(),
+                    autoUpdate,
+                    concreteIndices
+                ),
+                wrappedListener
+            )
+        );
     }
 
     static String checkForFailureStoreViolations(ClusterState clusterState, Index[] concreteIndices, PutMappingRequest request) {

server/src/main/java/org/elasticsearch/cluster/metadata/MetadataMappingService.java

@@ -56,7 +56,7 @@ public class MetadataMappingService {
     public MetadataMappingService(ClusterService clusterService, IndicesService indicesService) {
         this.clusterService = clusterService;
         this.indicesService = indicesService;
-        taskQueue = clusterService.createTaskQueue("put-mapping", Priority.HIGH, new PutMappingExecutor());
+        this.taskQueue = clusterService.createTaskQueue("put-mapping", Priority.HIGH, new PutMappingExecutor());
     }
 
     record PutMappingClusterStateUpdateTask(PutMappingClusterStateUpdateRequest request, ActionListener<AcknowledgedResponse> listener)

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/util/Holder.java

@@ -26,6 +26,16 @@ public void set(T value) {
         this.value = value;
     }
 
+    /**
+     * Sets a value in the holder, but only if none has already been set.
+     * @param value the new value to set.
+     */
+    public void setIfAbsent(T value) {
+        if (this.value == null) {
+            this.value = value;
+        }
+    }
+
     public T get() {
         return value;
     }

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java

@@ -11,6 +11,8 @@
 import org.apache.lucene.sandbox.document.HalfFloatPoint;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.breaker.CircuitBreaker;
 import org.elasticsearch.common.breaker.NoopCircuitBreaker;
@@ -31,9 +33,11 @@
 import org.elasticsearch.geo.ShapeTestUtils;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.search.SearchService;
 import org.elasticsearch.tasks.TaskCancelledException;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.transport.RemoteTransportException;
+import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xcontent.json.JsonXContent;
 import org.elasticsearch.xpack.esql.action.EsqlQueryResponse;
 import org.elasticsearch.xpack.esql.analysis.EnrichResolution;
@@ -72,8 +76,8 @@
 import org.elasticsearch.xpack.esql.plan.logical.local.LocalSupplier;
 import org.elasticsearch.xpack.esql.plugin.EsqlPlugin;
 import org.elasticsearch.xpack.esql.plugin.QueryPragmas;
+import org.elasticsearch.xpack.esql.plugin.TransportActionServices;
 import org.elasticsearch.xpack.esql.session.Configuration;
-import org.elasticsearch.xpack.esql.session.QueryBuilderResolver;
 import org.elasticsearch.xpack.esql.stats.SearchStats;
 import org.elasticsearch.xpack.esql.telemetry.Metrics;
 import org.elasticsearch.xpack.versionfield.Version;
@@ -140,6 +144,7 @@
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
 
 public final class EsqlTestUtils {
 
@@ -360,7 +365,14 @@ public static LogicalOptimizerContext unboundLogicalOptimizerContext() {
 
     public static final Verifier TEST_VERIFIER = new Verifier(new Metrics(new EsqlFunctionRegistry()), new XPackLicenseState(() -> 0L));
 
-    public static final QueryBuilderResolver MOCK_QUERY_BUILDER_RESOLVER = new MockQueryBuilderResolver();
+    public static final TransportActionServices MOCK_TRANSPORT_ACTION_SERVICES = new TransportActionServices(
+        mock(TransportService.class),
+        mock(SearchService.class),
+        null,
+        mock(ClusterService.class),
+        mock(IndexNameExpressionResolver.class),
+        null
+    );
 
     private EsqlTestUtils() {}
 

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/MockQueryBuilderResolver.java

@@ -246,7 +246,7 @@ public void testWhereMatchWithRow() {
         var error = expectThrows(ElasticsearchException.class, () -> run(query));
         assertThat(
             error.getMessage(),
-            containsString("[MATCH] function cannot operate on [\"a brown fox\"], which is not a field from an index mapping")
+            containsString("line 2:15: [MATCH] function cannot operate on [content], which is not a field from an index mapping")
         );
     }
 

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/plugin/MatchFunctionIT.java

@@ -230,7 +230,7 @@ public void testWhereMatchWithRow() {
         var error = expectThrows(ElasticsearchException.class, () -> run(query));
         assertThat(
             error.getMessage(),
-            containsString("[:] operator cannot operate on [\"a brown fox\"], which is not a field from an index mapping")
+            containsString("line 2:9: [:] operator cannot operate on [content], which is not a field from an index mapping")
         );
     }