Fix + test ROW | ENRICH

Author: alex-spies

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

@@ -556,12 +556,19 @@ public void testRowEnrich() throws IOException {
             String policyName = e.getKey() + "_policy";
             String query = "ROW " + LOOKUP_ID_FIELD + " = 123 | ENRICH " + policyName + " ON " + LOOKUP_ID_FIELD + " | LIMIT 1";
             var responseAndCoordinatorVersion = runQuery(query);
-            TransportVersion expectedMinimumVersion = minVersion(true);
-
-            assertMinimumVersion(expectedMinimumVersion, responseAndCoordinatorVersion);
-
             Map<String, Object> response = responseAndCoordinatorVersion.v1();
             TransportVersion coordinatorVersion = responseAndCoordinatorVersion.v2();
+            TransportVersion expectedMinimumVersion = minVersion(true);
+
+            Map<String, Object> profile = (Map<String, Object>) response.get("profile");
+            Integer actualMinimumVersion = (Integer) profile.get("minimumTransportVersion");
+            if (minVersion(true).supports(ESQL_USE_MINIMUM_VERSION_FOR_ENRICH_RESOLUTION)
+                // Some nodes don't send back the minimum transport version because they're too old to do that.
+                // In this case, the determined minimum version will be that of the coordinator.
+                || (coordinatorVersion.supports(ESQL_USE_MINIMUM_VERSION_FOR_ENRICH_RESOLUTION)
+                    && actualMinimumVersion != coordinatorVersion.id())) {
+                assertMinimumVersion(expectedMinimumVersion, responseAndCoordinatorVersion);
+            }
 
             assertNoPartialResponse(response);
 
@@ -645,17 +652,18 @@ protected void assertMinimumVersion(
             Map<String, Object> profile = (Map<String, Object>) responseMap.get("profile");
             Integer minimumVersion = (Integer) profile.get("minimumTransportVersion");
             assertNotNull(minimumVersion);
+            int minimumVersionInt = minimumVersion;
             if (expectedMinimumVersion.supports(RESOLVE_FIELDS_RESPONSE_CREATED_TV)) {
                 // All nodes are new enough that their field caps responses should contain the minimum transport version
                 // of matching clusters.
-                assertEquals(expectedMinimumVersion.id(), minimumVersion.intValue());
+                assertEquals(expectedMinimumVersion.id(), minimumVersionInt);
             } else {
                 // One node is old enough that it doesn't provide version information in the field caps response. We must assume
                 // the oldest compatible version.
                 // Since we got minimumVersion in the profile, the coordinator is on a new version.
                 // Thus, it's on the same version as this code (bwc tests only use 2 different versions) and the oldest compatible version
                 // to the coordinator is given by TransportVersion.minimumCompatible().
-                assertEquals(TransportVersion.minimumCompatible().id(), minimumVersion.intValue());
+                assertEquals(TransportVersion.minimumCompatible().id(), minimumVersionInt);
             }
         }
     }

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

@@ -42,13 +42,15 @@
 import org.elasticsearch.xpack.esql.analysis.EnrichResolution;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.type.EsField;
+import org.elasticsearch.xpack.esql.core.util.Holder;
 import org.elasticsearch.xpack.esql.core.util.StringUtils;
 import org.elasticsearch.xpack.esql.index.EsIndex;
 import org.elasticsearch.xpack.esql.index.IndexResolution;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.io.stream.PlanStreamOutput;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
 import org.elasticsearch.xpack.esql.session.IndexResolver;
+import org.elasticsearch.xpack.esql.session.Versioned;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -77,7 +79,6 @@
 public class EnrichPolicyResolver {
     private static final String RESOLVE_ACTION_NAME = "cluster:monitor/xpack/enrich/esql/resolve_policy";
 
-    // NOCOMMIT: rename to something that represents the overall change
     public static final TransportVersion ESQL_USE_MINIMUM_VERSION_FOR_ENRICH_RESOLUTION = TransportVersion.fromName(
         "esql_use_minimum_version_for_enrich_resolution"
     );
@@ -128,10 +129,10 @@ public void resolvePolicies(
         List<Enrich> enriches,
         EsqlExecutionInfo executionInfo,
         TransportVersion minimumVersion,
-        ActionListener<EnrichResolution> listener
+        ActionListener<Versioned<EnrichResolution>> listener
     ) {
         if (enriches.isEmpty()) {
-            listener.onResponse(new EnrichResolution());
+            listener.onResponse(new Versioned<>(new EnrichResolution(), minimumVersion));
             return;
         }
 
@@ -149,10 +150,10 @@ protected void doResolvePolicies(
         Collection<UnresolvedPolicy> unresolvedPolicies,
         EsqlExecutionInfo executionInfo,
         TransportVersion minimumVersion,
-        ActionListener<EnrichResolution> listener
+        ActionListener<Versioned<EnrichResolution>> listener
     ) {
         if (unresolvedPolicies.isEmpty()) {
-            listener.onResponse(new EnrichResolution());
+            listener.onResponse(new Versioned<>(new EnrichResolution(), minimumVersion));
             return;
         }
 
@@ -178,6 +179,14 @@ protected void doResolvePolicies(
                 }
             }
 
+            // Propagate the minimum version observed during policy resolution back to the planning pipeline.
+            // This is only really required for `ROW | ENRICH` queries, where the main index resolution doesn't
+            // provide the minimum version of the local cluster.
+            TransportVersion updatedMinimumVersion = minimumVersion;
+            for (LookupResponse response : lookupResponsesToProcess.values()) {
+                updatedMinimumVersion = TransportVersion.min(updatedMinimumVersion, response.minimumVersion);
+            }
+
             for (UnresolvedPolicy unresolved : unresolvedPolicies) {
                 Tuple<ResolvedEnrichPolicy, String> resolved = mergeLookupResults(
                     unresolved,
@@ -192,7 +201,7 @@ protected void doResolvePolicies(
                     enrichResolution.addError(unresolved.name, unresolved.mode, resolved.v2());
                 }
             }
-            return enrichResolution;
+            return new Versioned<>(enrichResolution, updatedMinimumVersion);
         }));
     }
 
@@ -404,6 +413,10 @@ private static class LookupRequest extends AbstractTransportRequest {
             if (in.getTransportVersion().supports(ESQL_USE_MINIMUM_VERSION_FOR_ENRICH_RESOLUTION)) {
                 this.minimumVersion = TransportVersion.readVersion(in);
             } else {
+                // An older coordinator contacted us. Let's assume an old version, otherwise we might send back
+                // types that it can't deserialize.
+                // (The only version that knows some new types but doesn't send its transport version here is 9.2.0;
+                // these types are dense_vector and aggregate_metric_double, and both don't work with ENRICH in 9.2.0, anyway.)
                 this.minimumVersion = TransportVersion.minimumCompatible();
             }
         }
@@ -421,12 +434,15 @@ public void writeTo(StreamOutput out) throws IOException {
     private static class LookupResponse extends TransportResponse {
         final Map<String, ResolvedEnrichPolicy> policies;
         final Map<String, String> failures;
+        // The minimum transport version observed when running field caps requests to resolve the policies
+        final TransportVersion minimumVersion;
         // does not need to be Writable since this indicates a failure to contact a remote cluster, so only set on querying cluster
         final transient Exception connectionError;
 
-        LookupResponse(Map<String, ResolvedEnrichPolicy> policies, Map<String, String> failures) {
+        LookupResponse(Map<String, ResolvedEnrichPolicy> policies, Map<String, String> failures, TransportVersion minimumVersion) {
             this.policies = policies;
             this.failures = failures;
+            this.minimumVersion = minimumVersion;
             this.connectionError = null;
         }
 
@@ -438,13 +454,27 @@ private static class LookupResponse extends TransportResponse {
         LookupResponse(Exception connectionError) {
             this.policies = Collections.emptyMap();
             this.failures = Collections.emptyMap();
+            this.minimumVersion = TransportVersion.current();
             this.connectionError = connectionError;
         }
 
         LookupResponse(StreamInput in) throws IOException {
             PlanStreamInput planIn = new PlanStreamInput(in, in.namedWriteableRegistry(), null);
             this.policies = planIn.readMap(StreamInput::readString, ResolvedEnrichPolicy::new);
             this.failures = planIn.readMap(StreamInput::readString, StreamInput::readString);
+            if (in.getTransportVersion().supports(ESQL_USE_MINIMUM_VERSION_FOR_ENRICH_RESOLUTION)) {
+                this.minimumVersion = TransportVersion.readVersion(in);
+            } else {
+                // A pre-9.2.1 node resolved the enrich policy for us, but doesn't say which version its cluster is on.
+                // We can safely assume this node's current version, even though that's technically wrong.
+                // Assuming a version that's too old can disable aggregate_metric_double and dense_vector
+                // data types in the query, that'd be very bad.
+                // But assuming these types are supported is fine because in 9.2.0,
+                // they're not supported in enrich policies, anyway, due to bugs.
+                // https://github.com/elastic/elasticsearch/issues/127350
+                // https://github.com/elastic/elasticsearch/issues/137699
+                this.minimumVersion = TransportVersion.minimumCompatible();
+            }
             this.connectionError = null;
         }
 
@@ -453,6 +483,9 @@ public void writeTo(StreamOutput out) throws IOException {
             PlanStreamOutput pso = new PlanStreamOutput(out, null);
             pso.writeMap(policies, StreamOutput::writeWriteable);
             pso.writeMap(failures, StreamOutput::writeString);
+            if (out.getTransportVersion().supports(ESQL_USE_MINIMUM_VERSION_FOR_ENRICH_RESOLUTION)) {
+                TransportVersion.writeVersion(minimumVersion, out);
+            }
         }
     }
 
@@ -462,12 +495,17 @@ public void messageReceived(LookupRequest request, TransportChannel channel, Tas
             final Map<String, EnrichPolicy> availablePolicies = availablePolicies();
             final Map<String, String> failures = ConcurrentCollections.newConcurrentMap();
             final Map<String, ResolvedEnrichPolicy> resolvedPolices = ConcurrentCollections.newConcurrentMap();
+            // We use the coordinator's minimum version as base line.
+            final Holder<TransportVersion> minimumVersion = new Holder<>(request.minimumVersion);
             ThreadContext threadContext = threadPool.getThreadContext();
             ActionListener<LookupResponse> listener = ContextPreservingActionListener.wrapPreservingContext(
                 new ChannelActionListener<>(channel),
                 threadContext
             );
-            try (var refs = new RefCountingListener(listener.map(unused -> new LookupResponse(resolvedPolices, failures)))) {
+            try (var refs = new RefCountingListener(listener.map(unused -> {
+                TransportVersion finalMinimumVersion = minimumVersion.get();
+                return new LookupResponse(resolvedPolices, failures, finalMinimumVersion);
+            }))) {
                 for (String policyName : request.policyNames) {
                     EnrichPolicy p = availablePolicies.get(policyName);
                     if (p == null) {
@@ -497,6 +535,11 @@ public void messageReceived(LookupRequest request, TransportChannel channel, Tas
                                         esIndex.mapping()
                                     );
                                     resolvedPolices.put(policyName, resolved);
+                                    synchronized (minimumVersion) {
+                                        minimumVersion.set(
+                                            TransportVersion.min(minimumVersion.get(), versionedIndexResult.minimumVersion())
+                                        );
+                                    }
                                 } else {
                                     failures.put(policyName, indexResult.toString());
                                 }

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

@@ -570,7 +570,10 @@ private void resolveIndicesAndAnalyze(
                     preAnalysis.enriches(),
                     executionInfo,
                     r.minimumTransportVersion(),
-                    l.map(r::withEnrichResolution)
+                    l.map(
+                        versionedResolution -> r.withEnrichResolution(versionedResolution.inner())
+                            .withMinimumTransportVersion(versionedResolution.minimumVersion())
+                    )
                 );
             })
             .<PreAnalysisResult>andThen((l, r) -> {
@@ -583,7 +586,9 @@ private void resolveIndicesAndAnalyze(
     }
 
     /**
-     * Perform a field caps request for each lookup index. Does not update the minimum transport version.
+     * Perform a field caps request for each lookup index.
+     * <p>
+     * Updates the minimum transport version to deal with ROW queries, where the main index resolution does not make a field caps request.
      */
     private void preAnalyzeLookupIndices(
         Iterator<IndexPattern> lookupIndices,

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

@@ -129,8 +129,12 @@ public void resolveAsMergedMapping(
                     useAggregateMetricDoubleWhenNotSupported,
                     useDenseVectorWhenNotSupported
                 );
-                // NOCOMMIT: Update logging here as well
-                LOGGER.debug("minimum transport version {} {}", response.caps().minTransportVersion(), info.effectiveMinTransportVersion());
+                LOGGER.debug(
+                    "updated minimum transport version from {} to effective version {} using version {} from field caps response",
+                    minimumVersion,
+                    info.effectiveMinTransportVersion(),
+                    response.caps().minTransportVersion()
+                );
                 l.onResponse(new Versioned<>(mergedMappings(indexPattern, info), info.effectiveMinTransportVersion()));
             })
         );

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

@@ -46,6 +46,7 @@
 import org.elasticsearch.xpack.esql.analysis.EnrichResolution;
 import org.elasticsearch.xpack.esql.plan.logical.Enrich;
 import org.elasticsearch.xpack.esql.session.IndexResolver;
+import org.elasticsearch.xpack.esql.session.Versioned;
 import org.junit.After;
 import org.junit.Before;
 
@@ -453,10 +454,9 @@ EnrichResolution resolvePolicies(Collection<String> clusters, Collection<Unresol
                     unresolvedPolicies.add(new UnresolvedPolicy("legacy-policy-1", randomFrom(Enrich.Mode.values())));
                 }
             }
-            PlainActionFuture<EnrichResolution> future = new PlainActionFuture<>();
-            // NOCOMMIT: We should pass in a sensible transport version in general and also test this with older ones.
+            PlainActionFuture<Versioned<EnrichResolution>> future = new PlainActionFuture<>();
             super.doResolvePolicies(new HashSet<>(clusters), unresolvedPolicies, esqlExecutionInfo, TransportVersion.current(), future);
-            return future.actionGet(30, TimeUnit.SECONDS);
+            return future.actionGet(30, TimeUnit.SECONDS).inner();
         }
 
         @Override

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;
@@ -82,8 +83,9 @@ EnrichPolicyResolver mockEnrichResolver() {
         EnrichPolicyResolver enrichResolver = mock(EnrichPolicyResolver.class);
         doAnswer(invocation -> {
             Object[] arguments = invocation.getArguments();
-            ActionListener<EnrichResolution> listener = (ActionListener<EnrichResolution>) arguments[arguments.length - 1];
-            listener.onResponse(new EnrichResolution());
+            ActionListener<Versioned<EnrichResolution>> listener = (ActionListener<Versioned<EnrichResolution>>) arguments[arguments.length
+                - 1];
+            listener.onResponse(new Versioned<>(new EnrichResolution(), TransportVersion.current()));
             return null;
         }).when(enrichResolver).resolvePolicies(any(), any(), any(), any());
         return enrichResolver;