ESQL: Enable new data types with created version (elastic#136327)

Enables `dense_vector` and `aggregate_metric_double` if all nodes in all
of the target clusters support it.

Relates to elastic#135193

Author: nik9000

docs/changelog/136327.yaml

@@ -0,0 +1,5 @@
+pr: 136327
+summary: Enable new data types with created version
+area: ES|QL
+type: enhancement
+issues: []

docs/reference/query-languages/esql/_snippets/functions/parameters/text_embedding.md

@@ -142,19 +142,6 @@ 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 {

docs/reference/query-languages/esql/images/functions/text_embedding.svg

@@ -7,6 +7,7 @@
 package org.elasticsearch.xpack.esql.core.type;
 
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.Build;
 import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -748,7 +749,7 @@ public DataType counter() {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        if (supportedVersion.supportedOn(out.getTransportVersion()) == false) {
+        if (supportedVersion.supportedOn(out.getTransportVersion(), Build.current().isSnapshot()) == 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

docs/reference/query-languages/esql/kibana/definition/functions/text_embedding.json

@@ -11,15 +11,15 @@
 import org.elasticsearch.TransportVersion;
 
 public interface SupportedVersion {
-    boolean supportedOn(TransportVersion version);
+    boolean supportedOn(TransportVersion version, boolean currentBuildIsSnapshot);
 
     default boolean supportedLocally() {
-        return supportedOn(TransportVersion.current());
+        return supportedOn(TransportVersion.current(), Build.current().isSnapshot());
     }
 
     SupportedVersion SUPPORTED_ON_ALL_NODES = new SupportedVersion() {
         @Override
-        public boolean supportedOn(TransportVersion version) {
+        public boolean supportedOn(TransportVersion version, boolean currentBuildIsSnapshot) {
             return true;
         }
 
@@ -56,8 +56,8 @@ public String toString() {
     // Check usage of this constant to be sure.
     SupportedVersion UNDER_CONSTRUCTION = new SupportedVersion() {
         @Override
-        public boolean supportedOn(TransportVersion version) {
-            return Build.current().isSnapshot();
+        public boolean supportedOn(TransportVersion version, boolean currentBuildIsSnapshot) {
+            return currentBuildIsSnapshot;
         }
 
         @Override
@@ -76,8 +76,8 @@ public String toString() {
     static SupportedVersion supportedSince(TransportVersion supportedVersion) {
         return new SupportedVersion() {
             @Override
-            public boolean supportedOn(TransportVersion version) {
-                return version.supports(supportedVersion) || Build.current().isSnapshot();
+            public boolean supportedOn(TransportVersion version, boolean currentBuildIsSnapshot) {
+                return version.supports(supportedVersion) || currentBuildIsSnapshot;
             }
 
             @Override

docs/reference/query-languages/esql/kibana/docs/functions/text_embedding.md

@@ -32,6 +32,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -43,6 +44,7 @@
 import static org.elasticsearch.test.ListMatcher.matchesList;
 import static org.elasticsearch.test.MapMatcher.assertMap;
 import static org.elasticsearch.test.MapMatcher.matchesMap;
+import static org.elasticsearch.xpack.esql.action.EsqlResolveFieldsResponse.RESOLVE_FIELDS_RESPONSE_CREATED_TV;
 import static org.hamcrest.Matchers.any;
 import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
@@ -76,11 +78,6 @@ public class AllSupportedFieldsTestCase extends ESRestTestCase {
 
     @ParametersFactory(argumentFormatting = "pref=%s mode=%s")
     public static List<Object[]> args() {
-        if (Build.current().isSnapshot()) {
-            // We only test behavior in release builds. Snapshot builds will have data types enabled that are still under construction.
-            return List.of();
-        }
-
         List<Object[]> args = new ArrayList<>();
         for (MappedFieldType.FieldExtractPreference extractPreference : Arrays.asList(
             null,
@@ -102,7 +99,7 @@ protected AllSupportedFieldsTestCase(MappedFieldType.FieldExtractPreference extr
         this.indexMode = indexMode;
     }
 
-    protected record NodeInfo(String cluster, String id, TransportVersion version, Set<String> roles) {}
+    protected record NodeInfo(String cluster, String id, boolean snapshot, TransportVersion version, Set<String> roles) {}
 
     private static Map<String, NodeInfo> nodeToInfo;
 
@@ -126,6 +123,19 @@ protected boolean fetchDenseVectorAggMetricDoubleIfFns() throws IOException {
         return clusterHasCapability("GET", "/_query", List.of(), List.of("DENSE_VECTOR_AGG_METRIC_DOUBLE_IF_FNS")).orElse(false);
     }
 
+    private static Boolean denseVectorAggMetricDoubleIfVersion;
+
+    private boolean denseVectorAggMetricDoubleIfVersion() throws IOException {
+        if (denseVectorAggMetricDoubleIfVersion == null) {
+            denseVectorAggMetricDoubleIfVersion = fetchDenseVectorAggMetricDoubleIfVersion();
+        }
+        return denseVectorAggMetricDoubleIfVersion;
+    }
+
+    protected boolean fetchDenseVectorAggMetricDoubleIfVersion() throws IOException {
+        return clusterHasCapability("GET", "/_query", List.of(), List.of("DENSE_VECTOR_AGG_METRIC_DOUBLE_IF_VERSION")).orElse(false);
+    }
+
     private static Boolean supportsNodeAssignment;
 
     protected boolean supportsNodeAssignment() throws IOException {
@@ -153,11 +163,21 @@ protected static Map<String, NodeInfo> fetchNodeToInfo(RestClient client, String
             String id = (String) n.getKey();
             Map<?, ?> nodeInfo = (Map<?, ?>) n.getValue();
             String nodeName = (String) extractValue(nodeInfo, "name");
+
+            /*
+             * Figuring out is a node is a snapshot is kind of tricky. The main version
+             * doesn't include -SNAPSHOT. But ${VERSION}-SNAPSHOT is in the node info
+             * *somewhere*. So we do this silly toString here.
+             */
+            String version = (String) extractValue(nodeInfo, "version");
+            boolean snapshot = nodeInfo.toString().contains(version + "-SNAPSHOT");
+
             TransportVersion transportVersion = TransportVersion.fromId((Integer) extractValue(nodeInfo, "transport_version"));
             List<?> roles = (List<?>) nodeInfo.get("roles");
+
             nodeToInfo.put(
                 nodeName,
-                new NodeInfo(cluster, id, transportVersion, roles.stream().map(Object::toString).collect(Collectors.toSet()))
+                new NodeInfo(cluster, id, snapshot, transportVersion, roles.stream().map(Object::toString).collect(Collectors.toSet()))
             );
         }
 
@@ -175,6 +195,22 @@ public void createIndices() throws IOException {
         }
     }
 
+    /**
+     * Make sure the test doesn't run on snapshot builds. Release builds only.
+     * <p>
+     *     {@link Build#isSnapshot()} checks if the version under test is a snapshot.
+     *     But! This run test runs against many versions and if *any* are snapshots
+     *     then this will fail. So we check the versions of each node in the cluster too.
+     * </p>
+     */
+    @Before
+    public void skipSnapshots() throws IOException {
+        assumeFalse("Only supported on production builds", Build.current().isSnapshot());
+        for (NodeInfo n : allNodeToInfo().values()) {
+            assumeFalse("Only supported on production builds", n.snapshot());
+        }
+    }
+
     // TODO: Also add a test for _tsid once we can determine the minimum transport version of all nodes.
     public final void testFetchAll() throws IOException {
         Map<String, Object> response = esql("""
@@ -212,7 +248,7 @@ public final void testFetchAll() throws IOException {
                 if (supportedInIndex(type) == false) {
                     continue;
                 }
-                expectedValues = expectedValues.entry(fieldName(type), expectedValue(type));
+                expectedValues = expectedValues.entry(fieldName(type), expectedValue(type, nodeInfo));
             }
             expectedValues = expectedValues.entry("_id", any(String.class))
                 .entry("_ignored", nullValue())
@@ -227,15 +263,23 @@ public final void testFetchAll() throws IOException {
         profileLogger.clearProfile();
     }
 
-    // Tests a workaround and will become obsolete once we can determine the actual minimum transport version of all nodes.
+    /**
+     * Tests fetching {@code dense_vector} if possible. Uses the {@code dense_vector_agg_metric_double_if_fns}
+     * work around if required.
+     */
     public final void testFetchDenseVector() throws IOException {
         Map<String, Object> response;
         try {
-            response = esql("""
-                | EVAL k = v_l2_norm(f_dense_vector, [1])  // workaround to enable fetching dense_vector
+            String request = """
                 | KEEP _index, f_dense_vector
                 | LIMIT 1000
-                """);
+                """;
+            if (denseVectorAggMetricDoubleIfVersion() == false) {
+                request = """
+                    | EVAL k = v_l2_norm(f_dense_vector, [1])  // workaround to enable fetching dense_vector
+                    """ + request;
+            }
+            response = esql(request);
             if ((Boolean) response.get("is_partial")) {
                 Map<?, ?> clusters = (Map<?, ?>) response.get("_clusters");
                 Map<?, ?> details = (Map<?, ?>) clusters.get("details");
@@ -410,7 +454,7 @@ private void createAllTypesDoc(RestClient client, String indexName) throws IOExc
     }
 
     // This will become dependent on the minimum transport version of all nodes once we can determine that.
-    private Matcher<?> expectedValue(DataType type) {
+    private Matcher<?> expectedValue(DataType type, NodeInfo nodeInfo) throws IOException {
         return switch (type) {
             case BOOLEAN -> equalTo(true);
             case COUNTER_LONG, LONG, COUNTER_INTEGER, INTEGER, UNSIGNED_LONG, SHORT, BYTE -> equalTo(1);
@@ -429,14 +473,24 @@ private Matcher<?> expectedValue(DataType type) {
             case GEO_SHAPE -> equalTo("POINT (-71.34 41.12)");
             case NULL -> nullValue();
             case AGGREGATE_METRIC_DOUBLE -> {
-                // Currently, we cannot tell if all nodes support it or not so we treat it as unsupported.
-                // TODO: Fix this once we know the node versions.
-                yield nullValue();
+                /*
+                 * We need both AGGREGATE_METRIC_DOUBLE_CREATED and RESOLVE_FIELDS_RESPONSE_CREATED_TV
+                 * but RESOLVE_FIELDS_RESPONSE_CREATED_TV came last so it's enough to check just it.
+                 */
+                if (minVersion().supports(RESOLVE_FIELDS_RESPONSE_CREATED_TV) == false) {
+                    yield nullValue();
+                }
+                yield equalTo("{\"min\":-302.5,\"max\":702.3,\"sum\":200.0,\"value_count\":25}");
             }
             case DENSE_VECTOR -> {
-                // Currently, we cannot tell if all nodes support it or not so we treat it as unsupported.
-                // TODO: Fix this once we know the node versions.
-                yield nullValue();
+                /*
+                 * We need both DENSE_VECTOR_CREATED and RESOLVE_FIELDS_RESPONSE_CREATED_TV
+                 * but RESOLVE_FIELDS_RESPONSE_CREATED_TV came last so it's enough to check just it.
+                 */
+                if (minVersion().supports(RESOLVE_FIELDS_RESPONSE_CREATED_TV) == false) {
+                    yield nullValue();
+                }
+                yield equalTo(List.of(0.5, 10.0, 5.9999995));
             }
             default -> throw new AssertionError("unsupported field type [" + type + "]");
         };
@@ -507,7 +561,7 @@ private Map<String, Object> nameToValue(List<String> names, List<?> values) {
     }
 
     // This will become dependent on the minimum transport version of all nodes once we can determine that.
-    private Matcher<String> expectedType(DataType type) {
+    private Matcher<String> expectedType(DataType type) throws IOException {
         return switch (type) {
             case COUNTER_DOUBLE, COUNTER_LONG, COUNTER_INTEGER -> {
                 if (indexMode == IndexMode.TIME_SERIES) {
@@ -518,10 +572,16 @@ private Matcher<String> expectedType(DataType type) {
             case BYTE, SHORT -> equalTo("integer");
             case HALF_FLOAT, SCALED_FLOAT, FLOAT -> equalTo("double");
             case NULL -> equalTo("keyword");
-            // Currently unsupported without TS command or KNN function
-            case AGGREGATE_METRIC_DOUBLE, DENSE_VECTOR ->
-                // TODO: Fix this once we know the node versions.
-                equalTo("unsupported");
+            case AGGREGATE_METRIC_DOUBLE, DENSE_VECTOR -> {
+                /*
+                 * We need both <type_name>_CREATED and RESOLVE_FIELDS_RESPONSE_CREATED_TV
+                 * but RESOLVE_FIELDS_RESPONSE_CREATED_TV came last so it's enough to check just it.
+                 */
+                if (minVersion().supports(RESOLVE_FIELDS_RESPONSE_CREATED_TV) == false) {
+                    yield equalTo("unsupported");
+                }
+                yield equalTo(type.esType());
+            }
             default -> equalTo(type.esType());
         };
     }
@@ -555,9 +615,13 @@ private Map<String, NodeInfo> expectedIndices() throws IOException {
                     name = e.getValue().cluster + ":" + name;
                 }
                 // We should only end up with one per cluster
-                result.put(name, new NodeInfo(e.getValue().cluster, null, e.getValue().version(), null));
+                result.put(name, new NodeInfo(e.getValue().cluster, null, e.getValue().snapshot(), e.getValue().version(), null));
             }
         }
         return result;
     }
+
+    protected TransportVersion minVersion() throws IOException {
+        return allNodeToInfo().values().stream().map(NodeInfo::version).min(Comparator.naturalOrder()).get();
+    }
 }

x-pack/plugin/build.gradle

@@ -1,9 +1,8 @@
 retrieveBitVectorData
-required_capability: dense_vector_field_type_released
+required_capability: dense_vector_agg_metric_double_if_version
 required_capability: l2_norm_vector_similarity_function
 
 FROM dense_vector
-| EVAL k = v_l2_norm(bit_vector, [1])  // workaround to enable fetching dense_vector
 | KEEP id, bit_vector
 | SORT id
 ;
@@ -16,11 +15,11 @@ id:l | bit_vector:dense_vector
 ;
 
 denseBitVectorWithEval
-required_capability: dense_vector_field_type_released
+required_capability: dense_vector_agg_metric_double_if_version
 required_capability: l2_norm_vector_similarity_function
 
 FROM dense_vector
-| EVAL v = bit_vector, k = v_l2_norm(bit_vector, [1])  // workaround to enable fetching dense_vector
+| EVAL v = bit_vector
 | KEEP id, v
 | SORT id
 ;
@@ -33,14 +32,13 @@ id:l | v:dense_vector
 ;
 
 denseBitVectorWithRenameAndDrop
-required_capability: dense_vector_field_type_released
+required_capability: dense_vector_agg_metric_double_if_version
 required_capability: l2_norm_vector_similarity_function
 
 FROM dense_vector
 | EVAL v = bit_vector
-| EVAL k = v_l2_norm(bit_vector, [1])  // workaround to enable fetching dense_vector
 | RENAME v AS new_vector
-| DROP float_vector, byte_vector, bit_vector, k
+| DROP float_vector, byte_vector, bit_vector
 | SORT id
 ;