ESQL: Fix enrich and lookup join resolution based on min transport version

docs/changelog/137431.yaml

@@ -0,0 +1,5 @@
+pr: 137431
+summary: Fix enrich and lookup join resolution based on min transport version
+area: ES|QL
+type: bug
+issues: []

server/src/main/resources/transport/definitions/referable/esql_use_minimum_version_for_enrich_resolution.csv

@@ -0,0 +1 @@
+9230000,9185011

server/src/main/resources/transport/upper_bounds/9.2.csv

@@ -1 +1 @@
-aggregate_metric_double_typed_block,9185010
+esql_use_minimum_version_for_enrich_resolution,9185011

server/src/main/resources/transport/upper_bounds/9.3.csv

@@ -1 +1 @@
-ingest_request_dynamic_template_params,9229000
+esql_use_minimum_version_for_enrich_resolution,9230000

test/framework/src/main/java/org/elasticsearch/test/rest/ESRestTestCase.java

@@ -2764,7 +2764,8 @@ protected static MapMatcher getProfileMatcher() {
             .entry("query", instanceOf(Map.class))
             .entry("planning", instanceOf(Map.class))
             .entry("drivers", instanceOf(List.class))
-            .entry("plans", instanceOf(List.class));
+            .entry("plans", instanceOf(List.class))
+            .entry("minimumTransportVersion", instanceOf(Integer.class));
     }
 
     protected static MapMatcher getResultMatcher(boolean includePartial, boolean includeDocumentsFound, boolean includeTimestamps) {

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

@@ -15,6 +15,7 @@
 import org.elasticsearch.index.IndexMode;
 import org.elasticsearch.index.mapper.SourceFieldMapper;
 import org.elasticsearch.index.mapper.TimeSeriesIdFieldMapper;
+import org.elasticsearch.xpack.esql.core.QlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.core.util.PlanStreamInput;
 import org.elasticsearch.xpack.esql.core.util.PlanStreamOutput;
 
@@ -759,13 +760,9 @@ public DataType counter() {
     public void writeTo(StreamOutput out) throws IOException {
         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
-             * 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 a 500 error - this is a bug, we failed to account for an old node during planning.
              */
-            throw new IllegalArgumentException(
+            throw new QlIllegalArgumentException(
                 "remote node at version [" + out.getTransportVersion() + "] doesn't understand data type [" + this + "]"
             );
         }

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

@@ -50,10 +50,10 @@ public AllSupportedFieldsIT(MappedFieldType.FieldExtractPreference extractPrefer
     public void createRemoteIndices() throws IOException {
         if (supportsNodeAssignment()) {
             for (Map.Entry<String, NodeInfo> e : remoteNodeToInfo().entrySet()) {
-                createIndexForNode(remoteClient(), e.getKey(), e.getValue().id());
+                createIndexForNode(remoteClient(), e.getKey(), e.getValue().id(), indexMode());
             }
         } else {
-            createIndexForNode(remoteClient(), null, null);
+            createIndexForNode(remoteClient(), null, null, indexMode());
         }
     }
 

x-pack/plugin/esql/qa/server/single-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/single_node/PushQueriesIT.java

@@ -366,6 +366,7 @@ private void testPushQuery(
                     .entry("plans", instanceOf(List.class))
                     .entry("planning", matchesMap().extraOk())
                     .entry("query", matchesMap().extraOk())
+                    .entry("minimumTransportVersion", instanceOf(Integer.class))
             ),
             matchesList().item(matchesMap().entry("name", "test").entry("type", anyOf(equalTo("text"), equalTo("keyword")))),
             equalTo(found ? List.of(List.of(value)) : List.of())

x-pack/plugin/esql/qa/server/single-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/single_node/StoredFieldsSequentialIT.java

@@ -122,6 +122,7 @@ private void testQuery(Double percent, String query, int documentsFound, boolean
                         .entry("plans", instanceOf(List.class))
                         .entry("planning", matchesMap().extraOk())
                         .entry("query", matchesMap().extraOk())
+                        .entry("minimumTransportVersion", instanceOf(Integer.class))
                 )
                 .extraOk()
         );