Fix index lookup when field-caps returns empty mapping

docs/changelog/132138.yaml

@@ -0,0 +1,6 @@
+pr: 132138
+summary: Fix lookup index resolution when field-caps returns empty mapping
+area: ES|QL
+type: bug
+issues:
+ - 132105

x-pack/plugin/build.gradle

@@ -139,6 +139,7 @@ tasks.named("yamlRestCompatTestTransform").configure({ task ->
   task.skipTest("esql/192_lookup_join_on_aliases/alias-pattern-multiple", "Error message changed")
   task.skipTest("esql/192_lookup_join_on_aliases/fails when alias or pattern resolves to multiple", "Error message changed")
   task.skipTest("esql/10_basic/Test wrong LIMIT parameter", "Error message changed")
+  task.skipTest("esql/190_lookup_join/lookup-no-key-only-key", "Requires the fix")
 })
 
 tasks.named('yamlRestCompatTest').configure {

x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterLookupJoinIT.java

@@ -10,6 +10,7 @@
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
 import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequestBuilder;
 import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.xpack.core.enrich.EnrichPolicy;
 import org.elasticsearch.xpack.core.enrich.action.ExecuteEnrichPolicyAction;
 import org.elasticsearch.xpack.core.enrich.action.PutEnrichPolicyAction;
@@ -289,6 +290,9 @@ public void testLookupJoinMissingKey() throws IOException {
         populateLookupIndex(REMOTE_CLUSTER_1, "values_lookup", 10);
 
         setSkipUnavailable(REMOTE_CLUSTER_1, true);
+
+        Exception ex;
+
         try (
             // Using local_tag as key which is not present in remote index
             EsqlQueryResponse resp = runQuery(
@@ -362,10 +366,7 @@ public void testLookupJoinMissingKey() throws IOException {
         }
 
         // TODO: verify whether this should be an error or not when the key field is missing
-        Exception ex = expectThrows(
-            VerificationException.class,
-            () -> runQuery("FROM c*:logs-* | LOOKUP JOIN values_lookup ON v", randomBoolean())
-        );
+        ex = expectThrows(VerificationException.class, () -> runQuery("FROM c*:logs-* | LOOKUP JOIN values_lookup ON v", randomBoolean()));
         assertThat(ex.getMessage(), containsString("Unknown column [v] in right side of join"));
 
         ex = expectThrows(
@@ -374,6 +375,25 @@ public void testLookupJoinMissingKey() throws IOException {
         );
         assertThat(ex.getMessage(), containsString("Unknown column [local_tag] in right side of join"));
 
+        // Add KEEP clause to try and trick the field-caps result parser into returning empty mapping
+        ex = expectThrows(
+            VerificationException.class,
+            () -> runQuery("FROM logs-* | LOOKUP JOIN values_lookup ON v | KEEP v", randomBoolean())
+        );
+        assertThat(ex.getMessage(), containsString("Unknown column [v] in right side of join"));
+
+        ex = expectThrows(
+            VerificationException.class,
+            () -> runQuery("FROM logs-*,c*:logs-* | LOOKUP JOIN values_lookup ON v | KEEP v", randomBoolean())
+        );
+        assertThat(ex.getMessage(), containsString("Unknown column [v] in right side of join"));
+
+        ex = expectThrows(
+            VerificationException.class,
+            () -> runQuery("FROM c*:logs-* | LOOKUP JOIN values_lookup ON v | KEEP v", randomBoolean())
+        );
+        assertThat(ex.getMessage(), containsString("Unknown column [v] in right side of join"));
+
         setSkipUnavailable(REMOTE_CLUSTER_1, false);
         try (
             // Using local_tag as key which is not present in remote index
@@ -393,6 +413,42 @@ public void testLookupJoinMissingKey() throws IOException {
             // FIXME: verify whether we need to succeed or fail here
             assertThat(remoteCluster.getStatus(), equalTo(EsqlExecutionInfo.Cluster.Status.SUCCESSFUL));
         }
+
+        // Add KEEP clause to try and trick the field-caps result parser into returning empty mapping
+        ex = expectThrows(
+            VerificationException.class,
+            () -> runQuery("FROM c*:logs-* | LOOKUP JOIN values_lookup ON v | KEEP v", randomBoolean())
+        );
+        assertThat(ex.getMessage(), containsString("Unknown column [v] in right side of join"));
+
+        ex = expectThrows(
+            VerificationException.class,
+            () -> runQuery("FROM logs-*,c*:logs-* | LOOKUP JOIN values_lookup ON v | KEEP v", randomBoolean())
+        );
+        assertThat(ex.getMessage(), containsString("Unknown column [v] in right side of join"));
+    }
+
+    public void testLookupJoinEmptyIndex() throws IOException {
+        setupClusters(2);
+        populateEmptyIndices(LOCAL_CLUSTER, "values_lookup");
+        populateEmptyIndices(REMOTE_CLUSTER_1, "values_lookup");
+
+        // Should work the same with both settings
+        setSkipUnavailable(REMOTE_CLUSTER_1, randomBoolean());
+
+        Exception ex;
+        for (String index : List.of("values_lookup", "values_lookup_map", "values_lookup_map_lookup")) {
+            ex = expectThrows(
+                VerificationException.class,
+                () -> runQuery("FROM logs-* | LOOKUP JOIN " + index + " ON v | KEEP v", randomBoolean())
+            );
+            assertThat(ex.getMessage(), containsString("Unknown column [v] in right side of join"));
+            ex = expectThrows(
+                VerificationException.class,
+                () -> runQuery("FROM c*:logs-* | LOOKUP JOIN " + index + " ON v | KEEP v", randomBoolean())
+            );
+            assertThat(ex.getMessage(), containsString("Unknown column [v] in right side of join"));
+        }
     }
 
     public void testLookupJoinIndexMode() throws IOException {
@@ -528,4 +584,23 @@ protected void setupAlias(String clusterAlias, String indexName, String aliasNam
         assertAcked(client.admin().indices().aliases(indicesAliasesRequestBuilder.request()));
     }
 
+    protected void populateEmptyIndices(String clusterAlias, String indexName) {
+        Client client = client(clusterAlias);
+        // Empty body
+        assertAcked(client.admin().indices().prepareCreate(indexName));
+        client.admin().indices().prepareRefresh(indexName).get();
+        // mappings + settings
+        assertAcked(
+            client.admin()
+                .indices()
+                .prepareCreate(indexName + "_map_lookup")
+                .setMapping()
+                .setSettings(Settings.builder().put("index.mode", "lookup"))
+        );
+        client.admin().indices().prepareRefresh(indexName + "_map_lookup").get();
+        // mappings only
+        assertAcked(client.admin().indices().prepareCreate(indexName + "_map").setMapping());
+        client.admin().indices().prepareRefresh(indexName + "_map").get();
+    }
+
 }

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

@@ -497,6 +497,11 @@ private PreAnalysisResult receiveLookupIndexResolution(
         }
         if (executionInfo.getClusters().isEmpty() || executionInfo.isCrossClusterSearch() == false) {
             // Local only case, still do some checks, since we moved analysis checks here
+            if (lookupIndexResolution.get().indexNameWithModes().isEmpty()) {
+                // This is not OK, but we proceed with it as we do with invalid resolution, and it will fail on the verification
+                // because lookup field will be missing.
+                return result.addLookupIndexResolution(index, lookupIndexResolution);
+            }
             if (lookupIndexResolution.get().indexNameWithModes().size() > 1) {
                 throw new VerificationException(
                     "Lookup Join requires a single lookup mode index; [" + index + "] resolves to multiple indices"
@@ -516,6 +521,16 @@ private PreAnalysisResult receiveLookupIndexResolution(
             }
             return result.addLookupIndexResolution(index, lookupIndexResolution);
         }
+
+        if (lookupIndexResolution.get().indexNameWithModes().isEmpty() && lookupIndexResolution.resolvedIndices().isEmpty() == false) {
+            // This is a weird situation - we have empty index list but non-empty resolution. This is likely because IndexResolver
+            // got an empty map and pretends to have an empty resolution. This means this query will fail, since lookup fields will not
+            // match, but here we can pretend it's ok to pass it on to the verifier and generate a correct error message.
+            // Note this only happens if the map is completely empty, which means it's going to error out anyway, since we should have
+            // at least the key field there.
+            return result.addLookupIndexResolution(index, lookupIndexResolution);
+        }
+
         // Collect resolved clusters from the index resolution, verify that each cluster has a single resolution for the lookup index
         Map<String, String> clustersWithResolvedIndices = new HashMap<>(lookupIndexResolution.resolvedIndices().size());
         lookupIndexResolution.get().indexNameWithModes().forEach((indexName, indexMode) -> {

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

@@ -157,6 +157,11 @@ public static IndexResolution mergedMappings(String indexPattern, FieldCapabilit
             allEmpty &= ir.get().isEmpty();
         }
         // If all the mappings are empty we return an empty set of resolved indices to line up with QL
+        // Introduced with #46775
+        // We need to be able to differentiate between an empty mapping index and an empty index due to fields not being found. An empty
+        // mapping index will generate no columns (important) for a query like FROM empty-mapping-index, whereas an empty result here but
+        // for fields that do not exist in the index (but the index has a mapping) will result in "VerificationException Unknown column"
+        // errors.
         var index = new EsIndex(indexPattern, rootFields, allEmpty ? Map.of() : concreteIndices, partiallyUnmappedFields);
         var failures = EsqlCCSUtils.groupFailuresPerCluster(fieldCapsResponse.getFailures());
         return IndexResolution.valid(index, concreteIndices.keySet(), failures);