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 @@
+9211000,9185007

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

@@ -1 +1 @@
-ilm_searchable_snapshot_opt_out_clone,9185006
+esql_use_minimum_version_for_enrich_resolution,9185007

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

@@ -1 +1 @@
-add_sample_method_downsample_ilm,9210000
+esql_use_minimum_version_for_enrich_resolution,9211000

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/AllSupportedFieldsTestCase.java

@@ -652,7 +652,6 @@ private Matcher<String> expectedType(DataType type) throws IOException {
                 yield equalTo("aggregate_metric_double");
             }
             case DENSE_VECTOR -> {
-                logger.error("ADFDAFAF " + minVersion());
                 if (minVersion().supports(RESOLVE_FIELDS_RESPONSE_USED_TV) == false
                     || minVersion().supports(ESQL_DENSE_VECTOR_CREATED_VERSION) == false) {
                     yield equalTo("unsupported");

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolver.java

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.enrich;
 
 import org.elasticsearch.ExceptionsHelper;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.support.ChannelActionListener;
@@ -75,6 +76,10 @@
 public class EnrichPolicyResolver {
     private static final String RESOLVE_ACTION_NAME = "cluster:monitor/xpack/enrich/esql/resolve_policy";
 
+    private static final TransportVersion ESQL_USE_MINIMUM_VERSION_FOR_ENRICH_RESOLUTION = TransportVersion.fromName(
+        "esql_use_minimum_version_for_enrich_resolution"
+    );
+
     private final ClusterService clusterService;
     private final IndexResolver indexResolver;
     private final TransportService transportService;
@@ -113,9 +118,16 @@ public static UnresolvedPolicy from(Enrich e) {
      *
      * @param enriches           the unresolved policies
      * @param executionInfo      the execution info
+     * @param minimumVersion     the minimum transport version of all clusters involved in the query, used for making the resolved mapping
+     *                           compatible with all possible nodes
      * @param listener           notified with the enrich resolution
      */
-    public void resolvePolicies(List<Enrich> enriches, EsqlExecutionInfo executionInfo, ActionListener<EnrichResolution> listener) {
+    public void resolvePolicies(
+        List<Enrich> enriches,
+        EsqlExecutionInfo executionInfo,
+        TransportVersion minimumVersion,
+        ActionListener<EnrichResolution> listener
+    ) {
         if (enriches.isEmpty()) {
             listener.onResponse(new EnrichResolution());
             return;
@@ -125,6 +137,7 @@ public void resolvePolicies(List<Enrich> enriches, EsqlExecutionInfo executionIn
             executionInfo.clusterInfo.isEmpty() ? new HashSet<>() : executionInfo.getRunningClusterAliases().collect(toSet()),
             enriches.stream().map(EnrichPolicyResolver.UnresolvedPolicy::from).toList(),
             executionInfo,
+            minimumVersion,
             listener
         );
     }
@@ -133,6 +146,7 @@ protected void doResolvePolicies(
         Set<String> remoteClusters,
         Collection<UnresolvedPolicy> unresolvedPolicies,
         EsqlExecutionInfo executionInfo,
+        TransportVersion minimumVersion,
         ActionListener<EnrichResolution> listener
     ) {
         if (unresolvedPolicies.isEmpty()) {
@@ -141,7 +155,7 @@ protected void doResolvePolicies(
         }
 
         final boolean includeLocal = remoteClusters.isEmpty() || remoteClusters.remove(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY);
-        lookupPolicies(remoteClusters, includeLocal, unresolvedPolicies, executionInfo, listener.map(lookupResponses -> {
+        lookupPolicies(remoteClusters, includeLocal, unresolvedPolicies, executionInfo, minimumVersion, listener.map(lookupResponses -> {
             final EnrichResolution enrichResolution = new EnrichResolution();
             final Map<String, LookupResponse> lookupResponsesToProcess = new HashMap<>();
             for (Map.Entry<String, LookupResponse> entry : lookupResponses.entrySet()) {
@@ -305,6 +319,7 @@ private void lookupPolicies(
         boolean includeLocal,
         Collection<UnresolvedPolicy> unresolvedPolicies,
         EsqlExecutionInfo executionInfo,
+        TransportVersion minimumVersion,
         ActionListener<Map<String, LookupResponse>> listener
     ) {
         final Map<String, LookupResponse> lookupResponses = ConcurrentCollections.newConcurrentMap();
@@ -324,7 +339,7 @@ public void onResponse(Transport.Connection connection) {
                             transportService.sendRequest(
                                 connection,
                                 RESOLVE_ACTION_NAME,
-                                new LookupRequest(cluster, remotePolicies),
+                                new LookupRequest(cluster, remotePolicies, minimumVersion),
                                 TransportRequestOptions.EMPTY,
                                 new ActionListenerResponseHandler<>(
                                     lookupListener.delegateResponse((l, e) -> failIfSkipUnavailableFalse(e, skipOnFailure, l)),
@@ -350,7 +365,7 @@ public void onFailure(Exception e) {
                 transportService.sendRequest(
                     transportService.getLocalNode(),
                     RESOLVE_ACTION_NAME,
-                    new LookupRequest(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY, localPolicies),
+                    new LookupRequest(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY, localPolicies, minimumVersion),
                     new ActionListenerResponseHandler<>(
                         refs.acquire(resp -> lookupResponses.put(RemoteClusterAware.LOCAL_CLUSTER_GROUP_KEY, resp)),
                         LookupResponse::new,
@@ -372,21 +387,32 @@ private void failIfSkipUnavailableFalse(Exception e, boolean skipOnFailure, Acti
     private static class LookupRequest extends AbstractTransportRequest {
         private final String clusterAlias;
         private final Collection<String> policyNames;
+        // The minimum version of all clusters involved in executing the ESQL query.
+        final TransportVersion minimumVersion;
 
-        LookupRequest(String clusterAlias, Collection<String> policyNames) {
+        LookupRequest(String clusterAlias, Collection<String> policyNames, TransportVersion minimumVersion) {
             this.clusterAlias = clusterAlias;
             this.policyNames = policyNames;
+            this.minimumVersion = minimumVersion;
         }
 
         LookupRequest(StreamInput in) throws IOException {
             this.clusterAlias = in.readString();
             this.policyNames = in.readStringCollectionAsList();
+            if (in.getTransportVersion().supports(ESQL_USE_MINIMUM_VERSION_FOR_ENRICH_RESOLUTION)) {
+                this.minimumVersion = TransportVersion.readVersion(in);
+            } else {
+                this.minimumVersion = TransportVersion.minimumCompatible();
+            }
         }
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeString(clusterAlias);
             out.writeStringCollection(policyNames);
+            if (out.getTransportVersion().supports(ESQL_USE_MINIMUM_VERSION_FOR_ENRICH_RESOLUTION)) {
+                TransportVersion.writeVersion(minimumVersion, out);
+            }
         }
     }
 
@@ -447,14 +473,15 @@ public void messageReceived(LookupRequest request, TransportChannel channel, Tas
                     }
                     try (ThreadContext.StoredContext ignored = threadContext.stashWithOrigin(ClientHelper.ENRICH_ORIGIN)) {
                         String indexName = EnrichPolicy.getBaseName(policyName);
-                        indexResolver.resolveAsMergedMapping(
+                        indexResolver.resolveAsMergedMappingForVersion(
                             indexName,
                             IndexResolver.ALL_FIELDS,
                             null,
                             false,
                             // Disable aggregate_metric_double and dense_vector until we get version checks in planning
                             false,
                             false,
+                            request.minimumVersion,
                             refs.acquire(indexResult -> {
                                 if (indexResult.isValid() && indexResult.get().concreteIndices().size() == 1) {
                                     EsIndex esIndex = indexResult.get();

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java

@@ -567,7 +567,12 @@ private void resolveIndicesAndAnalyze(
         })
             .<PreAnalysisResult>andThen((l, r) -> preAnalyzeLookupIndices(preAnalysis.lookupIndices().iterator(), r, executionInfo, l))
             .<PreAnalysisResult>andThen((l, r) -> {
-                enrichPolicyResolver.resolvePolicies(preAnalysis.enriches(), executionInfo, l.map(r::withEnrichResolution));
+                enrichPolicyResolver.resolvePolicies(
+                    preAnalysis.enriches(),
+                    executionInfo,
+                    r.minimumTransportVersion(),
+                    l.map(r::withEnrichResolution)
+                );
             })
             .<PreAnalysisResult>andThen((l, r) -> {
                 inferenceService.inferenceResolver(functionRegistry).resolveInferenceIds(parsed, l.map(r::withInferenceResolution));
@@ -610,14 +615,15 @@ private void preAnalyzeLookupIndex(
             ThreadPool.Names.SEARCH_COORDINATION,
             ThreadPool.Names.SYSTEM_READ
         );
-        indexResolver.resolveAsMergedMapping(
+        indexResolver.resolveAsMergedMappingForVersion(
             EsqlCCSUtils.createQualifiedLookupIndexExpressionFromAvailableClusters(executionInfo, localPattern),
             result.wildcardJoinIndices().contains(localPattern) ? IndexResolver.ALL_FIELDS : result.fieldNames,
             null,
             false,
             // Disable aggregate_metric_double and dense_vector until we get version checks in planning
             false,
             false,
+            result.minimumTransportVersion(),
             listener.map(indexResolution -> receiveLookupIndexResolution(result, localPattern, executionInfo, indexResolution))
         );
     }
@@ -851,7 +857,7 @@ private void preAnalyzeMainIndices(
                 result.withIndices(indexPattern, IndexResolution.valid(new EsIndex(indexPattern.indexPattern(), Map.of(), Map.of())))
             );
         } else {
-            indexResolver.resolveAsMergedMappingAndRetrieveMinimumVersion(
+            indexResolver.resolveAsMergedMapping(
                 indexPattern.indexPattern(),
                 result.fieldNames,
                 // Maybe if no indices are returned, retry without index mode and provide a clearer error message.

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/IndexResolver.java

@@ -84,58 +84,88 @@ public IndexResolver(Client client) {
     }
 
     /**
-     * Resolves a pattern to one (potentially compound meaning that spawns multiple indices) mapping.
+     * Like {@link #resolveAsMergedMapping(String, Set, QueryBuilder, boolean, boolean, boolean, ActionListener)},
+     * but uses a pre-determined minimum transport version.
      */
-    public void resolveAsMergedMapping(
-        String indexWildcard,
+    public void resolveAsMergedMappingForVersion(
+        String indexPattern,
         Set<String> fieldNames,
         QueryBuilder requestFilter,
         boolean includeAllDimensions,
         boolean useAggregateMetricDoubleWhenNotSupported,
         boolean useDenseVectorWhenNotSupported,
+        TransportVersion minimumVersion,
         ActionListener<IndexResolution> listener
     ) {
-        ActionListener<Versioned<IndexResolution>> ignoreVersion = listener.delegateFailureAndWrap(
+        ActionListener<Versioned<IndexResolution>> versionIgnoringListener = listener.delegateFailureAndWrap(
             (l, versionedResolution) -> l.onResponse(versionedResolution.inner())
         );
 
-        resolveAsMergedMappingAndRetrieveMinimumVersion(
-            indexWildcard,
+        resolveAsMergedMapping(
+            indexPattern,
             fieldNames,
             requestFilter,
             includeAllDimensions,
             useAggregateMetricDoubleWhenNotSupported,
             useDenseVectorWhenNotSupported,
-            ignoreVersion
+            minimumVersion,
+            versionIgnoringListener
         );
     }
 
     /**
-     * Resolves a pattern to one (potentially compound meaning that spawns multiple indices) mapping. Also retrieves the minimum transport
-     * version available in the cluster (and remotes).
+     * Perform a field caps request to resolve a pattern to one mapping (potentially compound, meaning it spans multiple indices).
+     * The field caps response contains the minimum transport version of all clusters that apply to the pattern,
+     * and it is used to deal with previously unsupported data types during resolution.
+     * <p>
+     * The retrieved minimum version is passed on to the listener.
+     * <p>
+     * If a field's type is not supported on the minimum version, it will be {@link DataType#UNSUPPORTED}.
      */
-    public void resolveAsMergedMappingAndRetrieveMinimumVersion(
-        String indexWildcard,
+    public void resolveAsMergedMapping(
+        String indexPattern,
+        Set<String> fieldNames,
+        QueryBuilder requestFilter,
+        boolean includeAllDimensions,
+        boolean useAggregateMetricDoubleWhenNotSupported,
+        boolean useDenseVectorWhenNotSupported,
+        ActionListener<Versioned<IndexResolution>> listener
+    ) {
+        resolveAsMergedMapping(
+            indexPattern,
+            fieldNames,
+            requestFilter,
+            includeAllDimensions,
+            useAggregateMetricDoubleWhenNotSupported,
+            useDenseVectorWhenNotSupported,
+            null,
+            listener
+        );
+    }
+
+    private void resolveAsMergedMapping(
+        String indexPattern,
         Set<String> fieldNames,
         QueryBuilder requestFilter,
         boolean includeAllDimensions,
         boolean useAggregateMetricDoubleWhenNotSupported,
         boolean useDenseVectorWhenNotSupported,
+        @Nullable TransportVersion minimumVersion,
         ActionListener<Versioned<IndexResolution>> listener
     ) {
         client.execute(
             EsqlResolveFieldsAction.TYPE,
-            createFieldCapsRequest(indexWildcard, fieldNames, requestFilter, includeAllDimensions),
+            createFieldCapsRequest(indexPattern, fieldNames, requestFilter, includeAllDimensions),
             listener.delegateFailureAndWrap((l, response) -> {
                 FieldsInfo info = new FieldsInfo(
                     response.caps(),
-                    response.caps().minTransportVersion(),
+                    minimumVersion == null ? response.caps().minTransportVersion() : minimumVersion,
                     Build.current().isSnapshot(),
                     useAggregateMetricDoubleWhenNotSupported,
                     useDenseVectorWhenNotSupported
                 );
                 LOGGER.debug("minimum transport version {} {}", response.caps().minTransportVersion(), info.effectiveMinTransportVersion());
-                l.onResponse(new Versioned<>(mergedMappings(indexWildcard, info), info.effectiveMinTransportVersion()));
+                l.onResponse(new Versioned<>(mergedMappings(indexPattern, info), info.effectiveMinTransportVersion()));
             })
         );
     }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/enrich/EnrichPolicyResolverTests.java

@@ -454,7 +454,8 @@ EnrichResolution resolvePolicies(Collection<String> clusters, Collection<Unresol
                 }
             }
             PlainActionFuture<EnrichResolution> future = new PlainActionFuture<>();
-            super.doResolvePolicies(new HashSet<>(clusters), unresolvedPolicies, esqlExecutionInfo, future);
+            // NOCOMMIT: We should pass in a sensible transport version in general and also test this with older ones.
+            super.doResolvePolicies(new HashSet<>(clusters), unresolvedPolicies, esqlExecutionInfo, TransportVersion.current(), future);
             return future.actionGet(30, TimeUnit.SECONDS);
         }
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/telemetry/PlanExecutorMetricsTests.java

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.telemetry;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.OriginalIndices;
 import org.elasticsearch.action.fieldcaps.FieldCapabilities;
@@ -84,7 +85,7 @@ EnrichPolicyResolver mockEnrichResolver() {
             ActionListener<EnrichResolution> listener = (ActionListener<EnrichResolution>) arguments[arguments.length - 1];
             listener.onResponse(new EnrichResolution());
             return null;
-        }).when(enrichResolver).resolvePolicies(any(), any(), any());
+        }).when(enrichResolver).resolvePolicies(any(), any(), TransportVersion.current(), any());
         return enrichResolver;
     }