ESQL: Opt in to support for new aggregate_metric_double and dense_vector using query constructs (elastic#135215)

Works around a problem with ESQL planning around new types. Without this
if you have either `aggregate_metric_double` or `dense_vector` fields
running on a node running < 9.2.0 ESQL will detect those for queries like
`FROM *` and request those fields. The old nodes don't know about the data
type. So they'll claim they can't parse the request and fail. That's bad
because in pure 9.1.x cluster `FROM *` would have returned `null` for
those fields rather than failing. And, come on, `FROM *` shouldn't fail
during an upgrade.

And it's not *just* during an upgrade. A mixed remote cluster being sent
`FROM remote:*` over cross cluster search will have the same problem if
it is 9.1 and the coordinating node is 9.2.

The fix we'd *like* to do is get the version of all nodes that'll have
the request and only enable the type if all versions support it. That's
complex, but it easy to think about and explain and handle in the planner.

We thought about a different fix - downgrading these fields to
`unsupported` on write to an old version - but that's difficult to reason
about and quite likely to fail in a long tail of weird ways.

This is something a simpler version of the "get all the versions and
disable unsupported fields" that uses a cute hack to not have to fetch the
version, which is the complex part. Instead, it scrapes the query for the
`TS` command or for functions like `KNN` or `TO_AGGREGATE_METRIC_DOUBLE`
and enables these fields if those are there. This works because folks who
want `dense_vector` will always be using `KNN` in 9.2.0. And folks who want `aggregate_metric_double` will use `TS`. These things will fail if the
entire cluster and all remote nodes aren't on 9.2.0 because they don't
exist on those versions. So users will tell us, by writing things that
require 9.2.0, if all the nodes are on 9.2.0.

We'll implement the transport version based approach in a follow up. But,
for now, we'll use this trick.

Fix elastic#135193

Author: nik9000

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
+            );
+    }
+}

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();
+    }
+}