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

When deciding which types are supported, we did not use the correct minimum transport version during the enrich resolution in case of CCS and ROW queries. What's more, the EnrichPolicyResolver did not account for the fact that the node requesting resolution might be on a version that doesn't support the types in the resolved mapping, which led to serialization bugs surfacing when trying to enable the DATE_RANGE type.

- Initialize the minimum transport version with the minimum version from the cluster state before any resolution steps. That makes ROW queries correct.
- Send the determined minimum transport version along the enrich resolution request so that remote clusters don't send un-deserializable data types back.
- Add the determined minimum transport version to the profile.
- Add a bunch of tests.

(cherry picked from commit 4a14f83)

# Conflicts:
#	server/src/main/resources/transport/upper_bounds/9.3.csv
#	x-pack/plugin/esql/qa/server/single-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/single_node/PushExpressionToLoadIT.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/TransportEsqlQueryAction.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java
#	x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/IndexResolver.java

Author: alex-spies

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