[9.2] ESQL: Fix enrich and lookup join resolution based on min transport version (#137431)

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 @@
+9231000,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 @@
-aggregate_metric_double_typed_block,9227000
+esql_use_minimum_version_for_enrich_resolution,9231000

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

@@ -2733,7 +2733,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) {

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;
 
@@ -751,13 +752,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,18 @@ 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());
+        }
+
+        // We need a single lookup index that has the same name across all clusters, as well as a single enrich policy per cluster.
+        // We create both only when we're testing LOOKUP mode.
+        if (indexExists(remoteClient(), LOOKUP_INDEX_NAME) == false && indexMode() == IndexMode.LOOKUP) {
+            createAllTypesIndex(remoteClient(), LOOKUP_INDEX_NAME, null, indexMode());
+            createAllTypesDoc(remoteClient(), LOOKUP_INDEX_NAME);
+            createEnrichPolicy(remoteClient(), LOOKUP_INDEX_NAME, ENRICH_POLICY_NAME);
         }
     }
 
@@ -101,4 +109,16 @@ && clusterHasCapability(remoteClient(), "GET", "/_query", List.of(), List.of("DE
                 false
             );
     }
+
+    @Override
+    protected boolean fetchAllIsCrossCluster() {
+        return true;
+    }
+
+    public final void testFetchAllOnlyFromRemotes() throws IOException {
+        doTestFetchAll(fromAllQuery("*:%mode%*", """
+            , _id, _ignored, _index_mode, _score, _source, _version
+            | LIMIT 1000
+            """), remoteNodeToInfo(), allNodeToInfo());
+    }
 }

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

@@ -365,6 +365,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()
         );