Merge branch 'main' into update_hkmeanstest

Author: john-wagster

server/src/main/java/org/elasticsearch/cluster/NotMasterException.java

@@ -9,12 +9,20 @@
 package org.elasticsearch.cluster;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+import org.elasticsearch.cluster.coordination.FailedToCommitClusterStateException;
 import org.elasticsearch.common.io.stream.StreamInput;
 
 import java.io.IOException;
 
 /**
- * Exception which indicates that an operation failed because the node stopped being the elected master.
+ * Exception indicating that a cluster state update operation failed because the node stopped being the elected master.
+ * Since this exception is thrown prior to the cluster state publication, it should only be used when the cluster state update
+ * <i>definitely</i> did not happen, and there is no possibility the next master committed the cluster state update.
+ *
+ * This is different from {@link FailedToCommitClusterStateException}.
+ *
+ * This exception is retryable within {@link TransportMasterNodeAction}.
  */
 public class NotMasterException extends ElasticsearchException {
 

server/src/main/java/org/elasticsearch/cluster/coordination/FailedToCommitClusterStateException.java

@@ -9,14 +9,20 @@
 package org.elasticsearch.cluster.coordination;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.action.support.master.TransportMasterNodeAction;
+import org.elasticsearch.cluster.NotMasterException;
 import org.elasticsearch.common.io.stream.StreamInput;
 
 import java.io.IOException;
 
 /**
  * Thrown when a cluster state publication fails to commit the new cluster state. If publication fails then a new master is elected but the
- * update might or might not take effect, depending on whether or not the newly-elected master accepted the published state that failed to
- * be committed.
+ * update might or might not take effect, depending on whether the newly-elected master accepted the published state that failed to
+ * be committed. This exception should only be used when there is <i>ambiguity</i> whether a state update took effect or not.
+ *
+ * This is different from {@link NotMasterException} where we know for certain that a state update never took effect.
+ *
+ * This exception is retryable within {@link TransportMasterNodeAction}.
  *
  * See {@link ClusterStatePublisher} for more details.
  */

x-pack/plugin/build.gradle

@@ -142,6 +142,19 @@ tasks.named("yamlRestCompatTestTransform").configure({ task ->
   task.skipTest("ml/sparse_vector_search/Search on a sparse_vector field with dots in the field names", "Vectors are no longer returned by default")
   task.skipTest("ml/sparse_vector_search/Search on a nested sparse_vector field with dots in the field names and conflicting child fields", "Vectors are no longer returned by default")
   task.skipTest("esql/190_lookup_join/lookup-no-key-only-key", "Requires the fix")
+  task.skipTest("esql/40_tsdb/aggregate_metric_double unsortable", "Extra function required to enable the field type")
+  task.skipTest("esql/40_tsdb/avg of aggregate_metric_double", "Extra function required to enable the field type")
+  task.skipTest("esql/40_tsdb/grouping stats on aggregate_metric_double", "Extra function required to enable the field type")
+  task.skipTest("esql/40_tsdb/render aggregate_metric_double when missing min and max", "Extra function required to enable the field type")
+  task.skipTest("esql/40_tsdb/render aggregate_metric_double when missing value", "Extra function required to enable the field type")
+  task.skipTest("esql/40_tsdb/sorting with aggregate_metric_double with partial submetrics", "Extra function required to enable the field type")
+  task.skipTest("esql/40_tsdb/stats on aggregate_metric_double missing min and max", "Extra function required to enable the field type")
+  task.skipTest("esql/40_tsdb/to_string aggregate_metric_double", "Extra function required to enable the field type")
+  task.skipTest("esql/40_tsdb/stats on aggregate_metric_double with partial submetrics", "Extra function required to enable the field type")
+  task.skipTest("esql/46_downsample/MV_EXPAND on non-MV aggregate metric double", "Extra function required to enable the field type")
+  task.skipTest("esql/46_downsample/Query stats on downsampled index", "Extra function required to enable the field type")
+  task.skipTest("esql/46_downsample/Render stats from downsampled index", "Extra function required to enable the field type")
+  task.skipTest("esql/46_downsample/Sort from multiple indices one with aggregate metric double", "Extra function required to enable the field type")
 })
 
 tasks.named('yamlRestCompatTest').configure {

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/CreatedVersion.java

@@ -0,0 +1,43 @@
+/*
+ * 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.esql.core.type;
+
+import org.elasticsearch.TransportVersion;
+
+/**
+ * Version that supports a {@link DataType}.
+ */
+public interface CreatedVersion {
+    boolean supports(TransportVersion version);
+
+    CreatedVersion SUPPORTED_ON_ALL_NODES = new CreatedVersion() {
+        @Override
+        public boolean supports(TransportVersion version) {
+            return true;
+        }
+
+        @Override
+        public String toString() {
+            return "SupportedOnAllVersions";
+        }
+    };
+
+    static CreatedVersion supportedOn(TransportVersion createdVersion) {
+        return new CreatedVersion() {
+            @Override
+            public boolean supports(TransportVersion version) {
+                return version.supports(createdVersion);
+            }
+
+            @Override
+            public String toString() {
+                return "SupportedOn[" + createdVersion + "]";
+            }
+        };
+    }
+}

x-pack/plugin/esql-core/src/main/java/org/elasticsearch/xpack/esql/core/type/DataType.java

@@ -7,8 +7,10 @@
 package org.elasticsearch.xpack.esql.core.type;
 
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.FeatureFlag;
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
@@ -32,6 +34,8 @@
 import java.util.function.Function;
 
 import static java.util.stream.Collectors.toMap;
+import static org.elasticsearch.TransportVersions.INFERENCE_REQUEST_ADAPTIVE_RATE_LIMITING;
+import static org.elasticsearch.TransportVersions.ML_INFERENCE_SAGEMAKER_CHAT_COMPLETION;
 
 /**
  * This enum represents data types the ES|QL query processing layer is able to
@@ -140,7 +144,7 @@
  *         unsupported types.</li>
  * </ul>
  */
-public enum DataType {
+public enum DataType implements Writeable {
     /**
      * Fields of this type are unsupported by any functions and are always
      * rendered as {@code null} in the response.
@@ -306,12 +310,26 @@ public enum DataType {
      */
     PARTIAL_AGG(builder().esType("partial_agg").estimatedSize(1024)),
 
-    AGGREGATE_METRIC_DOUBLE(builder().esType("aggregate_metric_double").estimatedSize(Double.BYTES * 3 + Integer.BYTES)),
+    AGGREGATE_METRIC_DOUBLE(
+        builder().esType("aggregate_metric_double")
+            .estimatedSize(Double.BYTES * 3 + Integer.BYTES)
+            .createdVersion(
+                // Version created just *after* we committed support for aggregate_metric_double
+                INFERENCE_REQUEST_ADAPTIVE_RATE_LIMITING
+            )
+    ),
 
     /**
      * Fields with this type are dense vectors, represented as an array of double values.
      */
-    DENSE_VECTOR(builder().esType("dense_vector").estimatedSize(4096));
+    DENSE_VECTOR(
+        builder().esType("dense_vector")
+            .estimatedSize(4096)
+            .createdVersion(
+                // Version created just *after* we committed support for dense_vector
+                ML_INFERENCE_SAGEMAKER_CHAT_COMPLETION
+            )
+    );
 
     /**
      * Types that are actively being built. These types are
@@ -375,6 +393,11 @@ public enum DataType {
      */
     private final DataType counter;
 
+    /**
+     * Version that first created this data type.
+     */
+    private final CreatedVersion createdVersion;
+
     DataType(Builder builder) {
         String typeString = builder.typeName != null ? builder.typeName : builder.esType;
         this.typeName = typeString.toLowerCase(Locale.ROOT);
@@ -387,6 +410,7 @@ public enum DataType {
         this.isCounter = builder.isCounter;
         this.widenSmallNumeric = builder.widenSmallNumeric;
         this.counter = builder.counter;
+        this.createdVersion = builder.createdVersion;
     }
 
     private static final Collection<DataType> TYPES = Arrays.stream(values())
@@ -727,7 +751,20 @@ public DataType counter() {
         return counter;
     }
 
+    @Override
     public void writeTo(StreamOutput out) throws IOException {
+        if (createdVersion.supports(out.getTransportVersion()) == false) {
+            /*
+             * TODO when we implement version aware planning flip this to an IllegalStateException
+             * so we throw a 500 error. It'll be our bug then. Right now it's a sign that the user
+             * tried to do something like `KNN(dense_vector_field, [1, 2])` against an old node.
+             * Like, during the rolling upgrade that enables KNN or to a remote cluster that has
+             * not yet been upgraded.
+             */
+            throw new IllegalArgumentException(
+                "remote node at version [" + out.getTransportVersion() + "] doesn't understand data type [" + this + "]"
+            );
+        }
         ((PlanStreamOutput) out).writeCachedString(typeName);
     }
 
@@ -779,6 +816,10 @@ public boolean isDate() {
         };
     }
 
+    public CreatedVersion createdVersion() {
+        return createdVersion;
+    }
+
     public static DataType suggestedCast(Set<DataType> originalTypes) {
         if (originalTypes.isEmpty() || originalTypes.contains(UNSUPPORTED)) {
             return null;
@@ -846,6 +887,13 @@ private static class Builder {
          */
         private DataType counter;
 
+        /**
+         * The version when this data type was created. We default to the first
+         * version for which we maintain wire compatibility, which is pretty
+         * much {@code 8.18.0}.
+         */
+        private CreatedVersion createdVersion = CreatedVersion.SUPPORTED_ON_ALL_NODES;
+
         Builder() {}
 
         Builder esType(String esType) {
@@ -901,5 +949,10 @@ Builder counter(DataType counter) {
             this.counter = counter;
             return this;
         }
+
+        Builder createdVersion(TransportVersion createdVersion) {
+            this.createdVersion = CreatedVersion.supportedOn(createdVersion);
+            return this;
+        }
     }
 }

x-pack/plugin/esql/qa/server/mixed-cluster/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/mixed/AllSupportedFieldsIT.java

@@ -0,0 +1,35 @@
+/*
+ * 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.esql.qa.mixed;
+
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
+
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.test.TestClustersThreadFilter;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.xpack.esql.qa.rest.AllSupportedFieldsTestCase;
+import org.junit.ClassRule;
+
+/**
+ * Fetch all field types in a mixed version cluster, simulating a rolling upgrade.
+ */
+@ThreadLeakFilters(filters = TestClustersThreadFilter.class)
+public class AllSupportedFieldsIT extends AllSupportedFieldsTestCase {
+    @ClassRule
+    public static ElasticsearchCluster cluster = Clusters.mixedVersionCluster();
+
+    public AllSupportedFieldsIT(MappedFieldType.FieldExtractPreference extractPreference, IndexMode indexMode) {
+        super(extractPreference, indexMode);
+    }
+
+    @Override
+    protected String getTestRestCluster() {
+        return cluster.getHttpAddresses();
+    }
+}

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/AllSupportedFieldsIT.java

@@ -0,0 +1,104 @@
+/*
+ * 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.esql.ccq;
+
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.core.IOUtils;
+import org.elasticsearch.index.IndexMode;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.test.TestClustersThreadFilter;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.xpack.esql.qa.rest.AllSupportedFieldsTestCase;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.rules.RuleChain;
+import org.junit.rules.TestRule;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * Fetch all field types via cross cluster search, possible on a different version.
+ */
+@ThreadLeakFilters(filters = TestClustersThreadFilter.class)
+public class AllSupportedFieldsIT extends AllSupportedFieldsTestCase {
+    static ElasticsearchCluster remoteCluster = Clusters.remoteCluster();
+    static ElasticsearchCluster localCluster = Clusters.localCluster(remoteCluster);
+
+    @ClassRule
+    public static TestRule clusterRule = RuleChain.outerRule(remoteCluster).around(localCluster);
+
+    private static RestClient remoteClient;
+    private static Map<String, NodeInfo> remoteNodeToInfo;
+
+    public AllSupportedFieldsIT(MappedFieldType.FieldExtractPreference extractPreference, IndexMode indexMode) {
+        super(extractPreference, indexMode);
+    }
+
+    @Before
+    public void createRemoteIndices() throws IOException {
+        if (supportsNodeAssignment()) {
+            for (Map.Entry<String, NodeInfo> e : remoteNodeToInfo().entrySet()) {
+                createIndexForNode(remoteClient(), e.getKey(), e.getValue().id());
+            }
+        } else {
+            createIndexForNode(remoteClient(), null, null);
+        }
+    }
+
+    private Map<String, NodeInfo> remoteNodeToInfo() throws IOException {
+        if (remoteNodeToInfo == null) {
+            remoteNodeToInfo = fetchNodeToInfo(remoteClient(), "remote_cluster");
+        }
+        return remoteNodeToInfo;
+    }
+
+    @Override
+    protected Map<String, NodeInfo> allNodeToInfo() throws IOException {
+        Map<String, NodeInfo> all = new TreeMap<>();
+        all.putAll(super.allNodeToInfo());
+        all.putAll(remoteNodeToInfo());
+        return all;
+    }
+
+    private RestClient remoteClient() throws IOException {
+        if (remoteClient == null) {
+            var clusterHosts = parseClusterHosts(remoteCluster.getHttpAddresses());
+            remoteClient = buildClient(restClientSettings(), clusterHosts.toArray(new HttpHost[0]));
+        }
+        return remoteClient;
+    }
+
+    @Override
+    protected String getTestRestCluster() {
+        return localCluster.getHttpAddresses();
+    }
+
+    @AfterClass
+    public static void closeRemoteClient() throws IOException {
+        try {
+            IOUtils.close(remoteClient);
+        } finally {
+            remoteClient = null;
+        }
+    }
+
+    @Override
+    protected boolean fetchDenseVectorAggMetricDoubleIfFns() throws IOException {
+        return super.fetchDenseVectorAggMetricDoubleIfFns()
+            && clusterHasCapability(remoteClient(), "GET", "/_query", List.of(), List.of("DENSE_VECTOR_AGG_METRIC_DOUBLE_IF_FNS")).orElse(
+                false
+            );
+    }
+}