Merge remote-tracking branch 'upstream/main' into refactor-xpack-deprecation-tests

ldematte · ldematte · commit 31dcf2f39287 · 2025-08-05T18:29:14.000+02:00
diff --git a/docs/changelog/132138.yaml b/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
diff --git a/muted-tests.yml b/muted-tests.yml
@@ -506,15 +506,6 @@ tests:
 - class: org.elasticsearch.xpack.ml.integration.AutodetectMemoryLimitIT
   method: testTooManyByAndOverFields
   issue: https://github.com/elastic/elasticsearch/issues/132310
-- class: org.elasticsearch.xpack.logsdb.qa.LogsDbVersusLogsDbReindexedIntoStandardModeChallengeRestIT
-  method: testTermsQuery
-  issue: https://github.com/elastic/elasticsearch/issues/132335
-- class: org.elasticsearch.xpack.logsdb.qa.LogsDbVersusReindexedIntoStoredSourceChallengeRestIT
-  method: testTermsQuery
-  issue: https://github.com/elastic/elasticsearch/issues/132336
-- class: org.elasticsearch.xpack.logsdb.qa.LogsDbVersusReindexedLogsDbChallengeRestIT
-  method: testTermsQuery
-  issue: https://github.com/elastic/elasticsearch/issues/132337
 - class: org.elasticsearch.xpack.esql.action.CrossClusterAsyncQueryIT
   method: testBadAsyncId
   issue: https://github.com/elastic/elasticsearch/issues/132353
diff --git a/server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java b/server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java
@@ -130,7 +130,7 @@ private void handleIncomingPublishRequest(
         ActionListener<PublishWithJoinResponse> publishResponseListener
     ) throws IOException {
         assert ThreadPool.assertCurrentThreadPool(GENERIC);
-        final Compressor compressor = CompressorFactory.compressor(request.bytes());
+        final Compressor compressor = CompressorFactory.compressorForUnknownXContentType(request.bytes());
         StreamInput in = request.bytes().streamInput();
         try {
             if (compressor != null) {
diff --git a/server/src/main/java/org/elasticsearch/common/compress/CompressedXContent.java b/server/src/main/java/org/elasticsearch/common/compress/CompressedXContent.java
@@ -135,7 +135,7 @@ public CompressedXContent(ToXContent xcontent, ToXContent.Params params) throws
      * that may already be compressed.
      */
     public CompressedXContent(BytesReference data) throws IOException {
-        Compressor compressor = CompressorFactory.compressor(data);
+        Compressor compressor = CompressorFactory.compressorForUnknownXContentType(data);
         if (compressor != null) {
             // already compressed...
             this.bytes = BytesReference.toBytes(data);
@@ -148,7 +148,7 @@ public CompressedXContent(BytesReference data) throws IOException {
     }
 
     private void assertConsistent() {
-        assert CompressorFactory.compressor(new BytesArray(bytes)) != null;
+        assert CompressorFactory.compressorForUnknownXContentType(new BytesArray(bytes)) != null;
         assert this.sha256.equals(sha256(uncompressed()));
         assert this.sha256.equals(sha256FromCompressed(bytes));
     }
diff --git a/server/src/main/java/org/elasticsearch/common/compress/CompressorFactory.java b/server/src/main/java/org/elasticsearch/common/compress/CompressorFactory.java
@@ -22,7 +22,7 @@ public class CompressorFactory {
     public static final Compressor COMPRESSOR = new DeflateCompressor();
 
     public static boolean isCompressed(BytesReference bytes) {
-        return compressor(bytes) != null;
+        return compressorForUnknownXContentType(bytes) != null;
     }
 
     @Nullable
@@ -34,6 +34,15 @@ public static Compressor compressor(BytesReference bytes) {
             assert XContentHelper.xContentType(bytes) == null;
             return COMPRESSOR;
         }
+        return null;
+    }
+
+    @Nullable
+    public static Compressor compressorForUnknownXContentType(BytesReference bytes) {
+        Compressor compressor = compressor(bytes);
+        if (compressor != null) {
+            return compressor;
+        }
 
         XContentType contentType = XContentHelper.xContentType(bytes);
         if (contentType == null) {
@@ -56,13 +65,13 @@ private static boolean isAncient(BytesReference bytes) {
      * @throws NullPointerException a NullPointerException will be thrown when bytes is null
      */
     public static BytesReference uncompressIfNeeded(BytesReference bytes) throws IOException {
-        Compressor compressor = compressor(Objects.requireNonNull(bytes, "the BytesReference must not be null"));
+        Compressor compressor = compressorForUnknownXContentType(Objects.requireNonNull(bytes, "the BytesReference must not be null"));
         return compressor == null ? bytes : compressor.uncompress(bytes);
     }
 
     /** Decompress the provided {@link BytesReference}. */
     public static BytesReference uncompress(BytesReference bytes) throws IOException {
-        Compressor compressor = compressor(bytes);
+        Compressor compressor = compressorForUnknownXContentType(bytes);
         if (compressor == null) {
             throw new NotCompressedException();
         }
diff --git a/server/src/main/java/org/elasticsearch/common/xcontent/XContentHelper.java b/server/src/main/java/org/elasticsearch/common/xcontent/XContentHelper.java
@@ -67,7 +67,7 @@ public static XContentParser createParser(NamedXContentRegistry registry, Deprec
      */
     @Deprecated
     public static XContentParser createParser(XContentParserConfiguration config, BytesReference bytes) throws IOException {
-        Compressor compressor = CompressorFactory.compressor(bytes);
+        Compressor compressor = CompressorFactory.compressorForUnknownXContentType(bytes);
         if (compressor != null) {
             InputStream compressedInput = compressor.threadLocalInputStream(bytes.streamInput());
             if (compressedInput.markSupported() == false) {
@@ -568,7 +568,7 @@ public interface CustomMerge {
     @Deprecated
     public static void writeRawField(String field, BytesReference source, XContentBuilder builder, ToXContent.Params params)
         throws IOException {
-        Compressor compressor = CompressorFactory.compressor(source);
+        Compressor compressor = CompressorFactory.compressorForUnknownXContentType(source);
         if (compressor != null) {
             try (InputStream compressedStreamInput = compressor.threadLocalInputStream(source.streamInput())) {
                 builder.rawField(field, compressedStreamInput);
@@ -592,7 +592,7 @@ public static void writeRawField(
         ToXContent.Params params
     ) throws IOException {
         Objects.requireNonNull(xContentType);
-        Compressor compressor = CompressorFactory.compressor(source);
+        Compressor compressor = CompressorFactory.compressorForUnknownXContentType(source);
         if (compressor != null) {
             try (InputStream compressedStreamInput = compressor.threadLocalInputStream(source.streamInput())) {
                 builder.rawField(field, compressedStreamInput, xContentType);
@@ -677,7 +677,7 @@ public static BytesReference toXContent(ChunkedToXContent toXContent, XContentTy
      */
     @Deprecated
     public static XContentType xContentTypeMayCompressed(BytesReference bytes) {
-        Compressor compressor = CompressorFactory.compressor(bytes);
+        Compressor compressor = CompressorFactory.compressorForUnknownXContentType(bytes);
         if (compressor != null) {
             try {
                 InputStream compressedStreamInput = compressor.threadLocalInputStream(bytes.streamInput());
diff --git a/server/src/test/java/org/elasticsearch/cluster/coordination/PublicationTransportHandlerTests.java b/server/src/test/java/org/elasticsearch/cluster/coordination/PublicationTransportHandlerTests.java
@@ -145,7 +145,7 @@ private static boolean isDiff(BytesTransportRequest request, TransportVersion ve
             StreamInput in = null;
             try {
                 in = request.bytes().streamInput();
-                final Compressor compressor = CompressorFactory.compressor(request.bytes());
+                final Compressor compressor = CompressorFactory.compressorForUnknownXContentType(request.bytes());
                 if (compressor != null) {
                     in = compressor.threadLocalStreamInput(in);
                 }
diff --git a/server/src/test/java/org/elasticsearch/common/xcontent/support/XContentHelperTests.java b/server/src/test/java/org/elasticsearch/common/xcontent/support/XContentHelperTests.java
@@ -12,12 +12,14 @@
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.compress.CompressedXContent;
+import org.elasticsearch.common.compress.NotXContentException;
 import org.elasticsearch.common.util.Maps;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.ToXContentObject;
 import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentParseException;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xcontent.XContentType;
@@ -421,4 +423,19 @@ public void testParseToType() throws IOException {
 
         assertThat(names, equalTo(Set.of("a", "c")));
     }
+
+    public void testGetParserWithInvalidInput() throws IOException {
+        assertThrows(
+            "Should detect bad JSON",
+            NotXContentException.class,
+            () -> XContentHelper.createParser(XContentParserConfiguration.EMPTY, new BytesArray("not actually XContent"))
+        );
+        XContentParser parser = XContentHelper.createParser(
+            XContentParserConfiguration.EMPTY,
+            new BytesArray("not actually XContent"),
+            XContentType.JSON
+        );
+        assertNotNull("Should not detect bad JSON", parser); // This is more like assertNotThrows
+        assertThrows("Should detect bad JSON at parse time", XContentParseException.class, parser::numberValue);
+    }
 }
diff --git a/x-pack/plugin/build.gradle b/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 {
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/CrossClusterLookupJoinIT.java b/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();
+    }
+
 }
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/EsqlSession.java b/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) -> {
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/session/IndexResolver.java b/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);
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/LogicalPlanOptimizerTests.java

Original file line number	Diff line number	Diff line change
`@@ -145,7 +145,7 @@ private static boolean isDiff(BytesTransportRequest request, TransportVersion ve`
`145`	`145`	`StreamInput in = null;`
`146`	`146`	`try {`
`147`	`147`	`in = request.bytes().streamInput();`
`148`		`- final Compressor compressor = CompressorFactory.compressor(request.bytes());`
	`148`	`+ final Compressor compressor = CompressorFactory.compressorForUnknownXContentType(request.bytes());`
`149`	`149`	`if (compressor != null) {`
`150`	`150`	`in = compressor.threadLocalStreamInput(in);`
`151`	`151`	`}`