ESQL: Opt in to support for new aggregate_metric_double and dense_vector using query constructs

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,91 @@
+/*
+ * 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.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 {
+        for (Map.Entry<String, NodeInfo> e : remoteNodeToInfo().entrySet()) {
+            createIndexForNode(remoteClient(), e.getKey(), e.getValue().id());
+        }
+    }
+
+    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;
+        }
+    }
+}

x-pack/plugin/esql/qa/server/single-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/single_node/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.single_node;
+
+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;
+
+/**
+ * Simple test for fetching all supported field types.
+ */
+@ThreadLeakFilters(filters = TestClustersThreadFilter.class)
+public class AllSupportedFieldsIT extends AllSupportedFieldsTestCase {
+    @ClassRule
+    public static ElasticsearchCluster cluster = Clusters.testCluster(c -> {});
+
+    public AllSupportedFieldsIT(MappedFieldType.FieldExtractPreference extractPreference, IndexMode indexMode) {
+        super(extractPreference, indexMode);
+    }
+
+    @Override
+    protected String getTestRestCluster() {
+        return cluster.getHttpAddresses();
+    }
+}