Update TransportVersion to support a new model

server/src/main/java/org/elasticsearch/TransportVersion.java

@@ -14,16 +14,28 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.internal.VersionExtension;
 import org.elasticsearch.plugins.ExtensionLoader;
+import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ParseField;
+import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentParserConfiguration;
+import org.elasticsearch.xcontent.json.JsonXContent;
 
+import java.io.BufferedReader;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.ServiceLoader;
 import java.util.function.Function;
+import java.util.function.IntFunction;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 /**
  * Represents the version of the wire protocol used to communicate between a pair of ES nodes.
@@ -57,7 +69,68 @@
  * different version value. If you need to know whether the cluster as a whole speaks a new enough {@link TransportVersion} to understand a
  * newly-added feature, use {@link org.elasticsearch.cluster.ClusterState#getMinTransportVersion}.
  */
-public record TransportVersion(int id) implements VersionId<TransportVersion> {
+public class TransportVersion implements VersionId<TransportVersion> {
+
+    private final String name;
+    private final int id;
+    private final TransportVersion nextPatchVersion;
+
+    public TransportVersion(int id) {
+        this(null, id, null);
+    }
+
+    public TransportVersion(String name, int id, TransportVersion patchVersion) {
+        this.name = name;
+        this.id = id;
+        this.nextPatchVersion = patchVersion;
+    }
+
+    public String name() {
+        return name;
+    }
+
+    public int id() {
+        return id;
+    }
+
+    public TransportVersion nextPatchVersion() {
+        return nextPatchVersion;
+    }
+
+    private static final ParseField NAME = new ParseField("name");
+    private static final ParseField IDS = new ParseField("ids");
+
+    private static final ConstructingObjectParser<TransportVersion, Integer> PARSER = new ConstructingObjectParser<>(
+        TransportVersion.class.getCanonicalName(),
+        false,
+        (args, latestTransportId) -> {
+            String name = (String) args[0];
+            @SuppressWarnings("unchecked")
+            List<Integer> ids = (List<Integer>) args[1];
+            ids.sort(Integer::compareTo);
+            TransportVersion transportVersion = null;
+            for (int idIndex = 0; idIndex < ids.size(); ++idIndex) {
+                if (ids.get(idIndex) > latestTransportId) {
+                    break;
+                }
+                transportVersion = new TransportVersion(name, ids.get(idIndex), transportVersion);
+            }
+            return transportVersion;
+        }
+    );
+
+    static {
+        PARSER.declareString(ConstructingObjectParser.constructorArg(), NAME);
+        PARSER.declareIntArray(ConstructingObjectParser.constructorArg(), IDS);
+    }
+
+    public static TransportVersion fromXContent(InputStream inputStream, Integer latest) {
+        try (XContentParser parser = JsonXContent.jsonXContent.createParser(XContentParserConfiguration.EMPTY, inputStream)) {
+            return PARSER.apply(parser, latest);
+        } catch (IOException ioe) {
+            throw new UncheckedIOException(ioe);
+        }
+    }
 
     public static TransportVersion readVersion(StreamInput in) throws IOException {
         return fromId(in.readVInt());
@@ -70,12 +143,20 @@ public static TransportVersion readVersion(StreamInput in) throws IOException {
      * The new instance is not registered in {@code TransportVersion.getAllVersions}.
      */
     public static TransportVersion fromId(int id) {
-        TransportVersion known = VersionsHolder.ALL_VERSIONS_MAP.get(id);
+        TransportVersion known = VersionsHolder.ALL_VERSIONS_BY_ID.get(id);
         if (known != null) {
             return known;
         }
         // this is a version we don't otherwise know about - just create a placeholder
-        return new TransportVersion(id);
+        return new TransportVersion(null, id, null);
+    }
+
+    public static TransportVersion fromName(String name) {
+        TransportVersion known = VersionsHolder.ALL_VERSIONS_BY_NAME.get(name);
+        if (known == null) {
+            throw new IllegalStateException("unknown transport version [" + name + "]");
+        }
+        return known;
     }
 
     public static void writeVersion(TransportVersion version, StreamOutput out) throws IOException {
@@ -123,7 +204,7 @@ public static List<TransportVersion> getAllVersions() {
      *         in the wild (they're sent over the wire by numeric ID) but we don't know how to communicate using such versions.
      */
     public boolean isKnown() {
-        return VersionsHolder.ALL_VERSIONS_MAP.containsKey(id);
+        return VersionsHolder.ALL_VERSIONS_BY_ID.containsKey(id);
     }
 
     /**
@@ -135,7 +216,7 @@ public TransportVersion bestKnownVersion() {
             return this;
         }
         TransportVersion bestSoFar = TransportVersions.ZERO;
-        for (final var knownVersion : VersionsHolder.ALL_VERSIONS_MAP.values()) {
+        for (final var knownVersion : VersionsHolder.ALL_VERSIONS_BY_ID.values()) {
             if (knownVersion.after(bestSoFar) && knownVersion.before(this)) {
                 bestSoFar = knownVersion;
             }
@@ -171,38 +252,134 @@ public boolean isPatchFrom(TransportVersion version) {
         return onOrAfter(version) && id < version.id + 100 - (version.id % 100);
     }
 
+    public boolean supports(TransportVersion version) {
+        if (onOrAfter(version)) {
+            return true;
+        }
+        TransportVersion nextPatchVersion = version.nextPatchVersion;
+        while (nextPatchVersion != null) {
+            if (isPatchFrom(version)) {
+                return true;
+            }
+            nextPatchVersion = nextPatchVersion.nextPatchVersion;
+        }
+        return false;
+    }
+
     /**
      * Returns a string representing the Elasticsearch release version of this transport version,
      * if applicable for this deployment, otherwise the raw version number.
      */
     public String toReleaseVersion() {
-        return TransportVersions.VERSION_LOOKUP.apply(id);
+        return VersionsHolder.VERSION_LOOKUP_BY_RELEASE.apply(id);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        TransportVersion that = (TransportVersion) o;
+        return id == that.id;
+    }
+
+    @Override
+    public int hashCode() {
+        return id;
     }
 
     @Override
     public String toString() {
-        return Integer.toString(id);
+        return "" + id;
     }
 
     private static class VersionsHolder {
+
         private static final List<TransportVersion> ALL_VERSIONS;
-        private static final Map<Integer, TransportVersion> ALL_VERSIONS_MAP;
+        private static final Map<Integer, TransportVersion> ALL_VERSIONS_BY_ID;
+        private static final Map<String, TransportVersion> ALL_VERSIONS_BY_NAME;
+        private static final IntFunction<String> VERSION_LOOKUP_BY_RELEASE;
         private static final TransportVersion CURRENT;
 
         static {
+            // collect all the transport versions from server and es modules/plugins (defined in server)
+            List<TransportVersion> allVersions = new ArrayList<>(TransportVersions.DEFINED_VERSIONS);
+            Map<String, TransportVersion> allVersionsByName = loadTransportVersionsByName();
+            addTransportVersions(allVersionsByName.values(), allVersions).sort(TransportVersion::compareTo);
+
+            // set version lookup by release before adding serverless versions
+            VERSION_LOOKUP_BY_RELEASE = ReleaseVersions.generateVersionsLookup(
+                TransportVersions.class,
+                allVersions.get(allVersions.size() - 1).id()
+            );
+
+            // collect all the transport versions from serverless
             Collection<TransportVersion> extendedVersions = ExtensionLoader.loadSingleton(ServiceLoader.load(VersionExtension.class))
                 .map(VersionExtension::getTransportVersions)
                 .orElse(Collections.emptyList());
+            addTransportVersions(extendedVersions, allVersions).sort(TransportVersion::compareTo);
+            for (TransportVersion version : extendedVersions) {
+                if (version.name() != null) {
+                    allVersionsByName.put(version.name(), version);
+                }
+            }
+
+            // set the transport version lookups
+            ALL_VERSIONS = Collections.unmodifiableList(allVersions);
+            ALL_VERSIONS_BY_ID = ALL_VERSIONS.stream().collect(Collectors.toUnmodifiableMap(TransportVersion::id, Function.identity()));
+            ALL_VERSIONS_BY_NAME = Collections.unmodifiableMap(allVersionsByName);
+            CURRENT = ALL_VERSIONS.getLast();
+        }
+
+        private static Map<String, TransportVersion> loadTransportVersionsByName() {
+            Map<String, TransportVersion> transportVersions = new HashMap<>();
+
+            String latestLocation = "/transport/latest/" + Version.CURRENT.major + "." + Version.CURRENT.minor + "-LATEST.json";
+            int latestId;
+            try (InputStream inputStream = TransportVersion.class.getResourceAsStream(latestLocation)) {
+                TransportVersion latest = fromXContent(inputStream, Integer.MAX_VALUE);
+                if (latest == null) {
+                    throw new IllegalStateException(
+                        "invalid latest transport version for release version [" + Version.CURRENT.major + "." + Version.CURRENT.minor + "]"
+                    );
+                }
+                latestId = latest.id();
+            } catch (IOException ioe) {
+                throw new UncheckedIOException("latest transport version file not found at [" + latestLocation + "]", ioe);
+            }
 
-            if (extendedVersions.isEmpty()) {
-                ALL_VERSIONS = TransportVersions.DEFINED_VERSIONS;
-            } else {
-                ALL_VERSIONS = Stream.concat(TransportVersions.DEFINED_VERSIONS.stream(), extendedVersions.stream()).sorted().toList();
+            String manifestLocation = "/transport/generated/generated-transport-versions-files-manifest.txt";
+            List<String> versionFileNames;
+            try (InputStream transportVersionsManifest = TransportVersion.class.getResourceAsStream(manifestLocation)) {
+                BufferedReader reader = new BufferedReader(new InputStreamReader(transportVersionsManifest, StandardCharsets.UTF_8));
+                versionFileNames = reader.lines().filter(line -> line.isBlank() == false).toList();
+            } catch (IOException ioe) {
+                throw new UncheckedIOException("transport version manifest file not found at [" + manifestLocation + "]", ioe);
             }
 
-            ALL_VERSIONS_MAP = ALL_VERSIONS.stream().collect(Collectors.toUnmodifiableMap(TransportVersion::id, Function.identity()));
+            for (String name : versionFileNames) {
+                String versionLocation = "/transport/generated/" + name;
+                try (InputStream inputStream = TransportVersion.class.getResourceAsStream(versionLocation)) {
+                    TransportVersion transportVersion = TransportVersion.fromXContent(inputStream, latestId);
+                    if (transportVersion != null) {
+                        transportVersions.put(transportVersion.name(), transportVersion);
+                    }
+                } catch (IOException ioe) {
+                    throw new UncheckedIOException("transport version set file not found at [ " + versionLocation + "]", ioe);
+                }
+            }
 
-            CURRENT = ALL_VERSIONS.getLast();
+            return transportVersions;
+        }
+
+        private static List<TransportVersion> addTransportVersions(Collection<TransportVersion> addFrom, List<TransportVersion> addTo) {
+            for (TransportVersion transportVersion : addFrom) {
+                addTo.add(transportVersion);
+                TransportVersion patchVersion = transportVersion.nextPatchVersion();
+                while (patchVersion != null) {
+                    addTo.add(patchVersion);
+                    patchVersion = patchVersion.nextPatchVersion();
+                }
+            }
+            return addTo;
         }
     }
 }

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -19,7 +19,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeSet;
-import java.util.function.IntFunction;
 
 /**
  * <p>Transport version is used to coordinate compatible wire protocol communication between nodes, at a fine-grained level.  This replaces
@@ -210,7 +209,6 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_COHERE_API_VERSION_8_19 = def(8_841_0_60);
     public static final TransportVersion ESQL_DOCUMENTS_FOUND_AND_VALUES_LOADED_8_19 = def(8_841_0_61);
     public static final TransportVersion ESQL_PROFILE_INCLUDE_PLAN_8_19 = def(8_841_0_62);
-    public static final TransportVersion ESQL_SPLIT_ON_BIG_VALUES_8_19 = def(8_841_0_63);
     public static final TransportVersion ESQL_FIXED_INDEX_LIKE_8_19 = def(8_841_0_64);
     public static final TransportVersion V_9_0_0 = def(9_000_0_09);
     public static final TransportVersion INITIAL_ELASTICSEARCH_9_0_1 = def(9_000_0_10);
@@ -328,26 +326,22 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_COHERE_API_VERSION = def(9_110_0_00);
     public static final TransportVersion ESQL_PROFILE_INCLUDE_PLAN = def(9_111_0_00);
     public static final TransportVersion MAPPINGS_IN_DATA_STREAMS = def(9_112_0_00);
-    public static final TransportVersion ESQL_SPLIT_ON_BIG_VALUES_9_1 = def(9_112_0_01);
     public static final TransportVersion ESQL_FIXED_INDEX_LIKE_9_1 = def(9_112_0_02);
     public static final TransportVersion ESQL_SAMPLE_OPERATOR_STATUS_9_1 = def(9_112_0_03);
     // Below is the first version in 9.2 and NOT in 9.1.
     public static final TransportVersion PROJECT_STATE_REGISTRY_RECORDS_DELETIONS = def(9_113_0_00);
     public static final TransportVersion ESQL_SERIALIZE_TIMESERIES_FIELD_TYPE = def(9_114_0_00);
     public static final TransportVersion ML_INFERENCE_IBM_WATSONX_COMPLETION_ADDED = def(9_115_0_00);
-    public static final TransportVersion ESQL_SPLIT_ON_BIG_VALUES = def(9_116_0_00);
     public static final TransportVersion ESQL_LOCAL_RELATION_WITH_NEW_BLOCKS = def(9_117_0_00);
     public static final TransportVersion ML_INFERENCE_CUSTOM_SERVICE_EMBEDDING_TYPE = def(9_118_0_00);
     public static final TransportVersion ESQL_FIXED_INDEX_LIKE = def(9_119_0_00);
     public static final TransportVersion LOOKUP_JOIN_CCS = def(9_120_0_00);
     public static final TransportVersion NODE_USAGE_STATS_FOR_THREAD_POOLS_IN_CLUSTER_INFO = def(9_121_0_00);
     public static final TransportVersion ESQL_CATEGORIZE_OPTIONS = def(9_122_0_00);
-    public static final TransportVersion ML_INFERENCE_AZURE_AI_STUDIO_RERANK_ADDED = def(9_123_0_00);
     public static final TransportVersion PROJECT_STATE_REGISTRY_ENTRY = def(9_124_0_00);
     public static final TransportVersion ML_INFERENCE_LLAMA_ADDED = def(9_125_0_00);
     public static final TransportVersion SHARD_WRITE_LOAD_IN_CLUSTER_INFO = def(9_126_0_00);
     public static final TransportVersion ESQL_SAMPLE_OPERATOR_STATUS = def(9_127_0_00);
-    public static final TransportVersion ESQL_TOPN_TIMINGS = def(9_128_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,
@@ -421,16 +415,6 @@ static TransportVersion def(int id) {
      */
     static final List<TransportVersion> DEFINED_VERSIONS = collectAllVersionIdsDefinedInClass(TransportVersions.class);
 
-    // the highest transport version constant defined
-    static final TransportVersion LATEST_DEFINED;
-    static {
-        LATEST_DEFINED = DEFINED_VERSIONS.getLast();
-
-        // see comment on IDS field
-        // now we're registered all the transport versions, we can clear the map
-        IDS = null;
-    }
-
     public static List<TransportVersion> collectAllVersionIdsDefinedInClass(Class<?> cls) {
         Map<Integer, String> versionIdFields = new HashMap<>();
         List<TransportVersion> definedTransportVersions = new ArrayList<>();
@@ -472,8 +456,6 @@ public static List<TransportVersion> collectAllVersionIdsDefinedInClass(Class<?>
         return List.copyOf(definedTransportVersions);
     }
 
-    static final IntFunction<String> VERSION_LOOKUP = ReleaseVersions.generateVersionsLookup(TransportVersions.class, LATEST_DEFINED.id());
-
     // no instance
     private TransportVersions() {}
 }

server/src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -76,12 +76,15 @@
  */
 public abstract class StreamInput extends InputStream {
 
-    private TransportVersion version = TransportVersion.current();
+    private TransportVersion version;
 
     /**
      * The transport version the data is serialized as.
      */
     public TransportVersion getTransportVersion() {
+        if (version == null) {
+            version = TransportVersion.current();
+        }
         return this.version;
     }
 

server/src/main/resources/transport/generated/esql-split-on-big-values.json

@@ -0,0 +1,8 @@
+{
+  "name": "esql-split-on-big-values",
+  "ids": [
+    9116000,
+    9112001,
+    8841063
+  ]
+}

server/src/main/resources/transport/generated/esql-topn-timings.json

@@ -0,0 +1,6 @@
+{
+  "name": "esql-topn-timings",
+  "ids": [
+    9128000
+  ]
+}

server/src/main/resources/transport/generated/generated-transport-versions-files-manifest.txt

@@ -0,0 +1,3 @@
+esql-split-on-big-values.json
+ml-inference-azure-ai-studio-rerank-added.json
+esql-topn-timings.json

server/src/main/resources/transport/generated/ml-inference-azure-ai-studio-rerank-added.json

@@ -0,0 +1,6 @@
+{
+  "name": "ml-inference-azure-ai-studio-rerank-added",
+  "ids": [
+    9123000
+  ]
+}