Remote Lookup Join implementation (#129013)

Remote lookup join implementation

This patch enables using LOOKUP JOIN with cross-cluster queries. Example:
FROM logs-*, remote:logs-* | LOOKUP JOIN clients on ip | SORT timestamp | LIMIT 100

Author: smalyshev

docs/changelog/129013.yaml

@@ -0,0 +1,13 @@
+pr: 129013
+summary: "Add remote index support to LOOKUP JOIN"
+area: ES|QL
+type: feature
+issues: [ ]
+highlight:
+  title: Add remote index support to LOOKUP JOIN
+  body: |-
+    Queries containing LOOKUP JOIN now can be preformed on cross-cluster indices, for example:
+    [source,yaml]
+    ----------------------------
+    FROM logs-*, remote:logs-* | LOOKUP JOIN clients on ip | SORT timestamp | LIMIT 100
+    ----------------------------

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -336,6 +336,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_LOCAL_RELATION_WITH_NEW_BLOCKS = def(9_117_0_00);
     public static final TransportVersion ML_INFERENCE_CUSTOM_SERVICE_EMBEDDING_TYPE = def(9_118_0_00);
     public static final TransportVersion ESQL_FIXED_INDEX_LIKE = def(9_119_0_00);
+    public static final TransportVersion LOOKUP_JOIN_CCS = def(9_120_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/build.gradle

@@ -136,6 +136,8 @@ tasks.named("yamlRestCompatTestTransform").configure({ task ->
   task.skipTest("esql/191_lookup_join_on_datastreams/data streams not supported in LOOKUP JOIN", "Added support for aliases in JOINs")
   task.skipTest("esql/190_lookup_join/non-lookup index", "Error message changed")
   task.skipTest("esql/192_lookup_join_on_aliases/alias-pattern-multiple", "Error message changed")
+  task.skipTest("esql/190_lookup_join/fails with non-lookup index", "Error message changed")
+  task.skipTest("esql/192_lookup_join_on_aliases/fails when alias or pattern resolves to multiple", "Error message changed")
 })
 
 tasks.named('yamlRestCompatTest').configure {

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterSpecIT.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.test.rest.TestFeatureService;
 import org.elasticsearch.xpack.esql.CsvSpecReader;
 import org.elasticsearch.xpack.esql.CsvSpecReader.CsvTestCase;
+import org.elasticsearch.xpack.esql.CsvTestsDataLoader;
 import org.elasticsearch.xpack.esql.SpecReader;
 import org.elasticsearch.xpack.esql.qa.rest.EsqlSpecTestCase;
 import org.junit.AfterClass;
@@ -39,13 +40,16 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Optional;
+import java.util.Set;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.xpack.esql.CsvSpecReader.specParser;
 import static org.elasticsearch.xpack.esql.CsvTestUtils.isEnabled;
+import static org.elasticsearch.xpack.esql.CsvTestsDataLoader.CSV_DATASET_MAP;
 import static org.elasticsearch.xpack.esql.CsvTestsDataLoader.ENRICH_SOURCE_INDICES;
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.classpathResources;
+import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.ENABLE_LOOKUP_JOIN_ON_REMOTE;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.FORK_V9;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.INLINESTATS;
 import static org.elasticsearch.xpack.esql.action.EsqlCapabilities.Cap.INLINESTATS_V2;
@@ -109,6 +113,22 @@ public MultiClusterSpecIT(
         super(fileName, groupName, testName, lineNumber, convertToRemoteIndices(testCase), instructions, mode);
     }
 
+    // TODO: think how to handle this better
+    public static final Set<String> NO_REMOTE_LOOKUP_JOIN_TESTS = Set.of(
+        // Lookup join after STATS is not supported in CCS yet
+        "StatsAndLookupIPAndMessageFromIndex",
+        "JoinMaskingRegex",
+        "StatsAndLookupIPFromIndex",
+        "StatsAndLookupMessageFromIndex",
+        "MvJoinKeyOnTheLookupIndexAfterStats",
+        "MvJoinKeyOnFromAfterStats",
+        // Lookup join after SORT is not supported in CCS yet
+        "NullifiedJoinKeyToPurgeTheJoin",
+        "SortBeforeAndAfterJoin",
+        "SortEvalBeforeLookup",
+        "SortBeforeAndAfterMultipleJoinAndMvExpand"
+    );
+
     @Override
     protected void shouldSkipTest(String testName) throws IOException {
         boolean remoteMetadata = testCase.requiredCapabilities.contains(METADATA_FIELDS_REMOTE_TEST.capabilityName());
@@ -129,10 +149,20 @@ protected void shouldSkipTest(String testName) throws IOException {
         assumeFalse("INLINESTATS not yet supported in CCS", testCase.requiredCapabilities.contains(INLINESTATS_V2.capabilityName()));
         assumeFalse("INLINESTATS not yet supported in CCS", testCase.requiredCapabilities.contains(JOIN_PLANNING_V1.capabilityName()));
         assumeFalse("INLINESTATS not yet supported in CCS", testCase.requiredCapabilities.contains(INLINESTATS_V8.capabilityName()));
-        assumeFalse("LOOKUP JOIN not yet supported in CCS", testCase.requiredCapabilities.contains(JOIN_LOOKUP_V12.capabilityName()));
+        if (testCase.requiredCapabilities.contains(JOIN_LOOKUP_V12.capabilityName())) {
+            assumeTrue("LOOKUP JOIN not yet supported in CCS", hasCapabilities(List.of(ENABLE_LOOKUP_JOIN_ON_REMOTE.capabilityName())));
+        }
         // Unmapped fields require a coorect capability response from every cluster, which isn't currently implemented.
         assumeFalse("UNMAPPED FIELDS not yet supported in CCS", testCase.requiredCapabilities.contains(UNMAPPED_FIELDS.capabilityName()));
         assumeFalse("FORK not yet supported in CCS", testCase.requiredCapabilities.contains(FORK_V9.capabilityName()));
+        // Tests that use capabilities not supported in CCS
+        assumeFalse(
+            "This syntax is not supported with remote LOOKUP JOIN",
+            NO_REMOTE_LOOKUP_JOIN_TESTS.stream().anyMatch(testName::contains)
+        );
+        // Tests that do SORT before LOOKUP JOIN - not supported in CCS
+        assumeFalse("LOOKUP JOIN after SORT not yet supported in CCS", testName.contains("OnTheCoordinator"));
+
     }
 
     @Override
@@ -181,6 +211,19 @@ protected RestClient buildClient(Settings settings, HttpHost[] localHosts) throw
     // These indices are used in metadata tests so we want them on remote only for consistency
     public static final List<String> METADATA_INDICES = List.of("employees", "apps", "ul_logs");
 
+    // These are lookup indices, we want them on both remotes and locals
+    public static final Set<String> LOOKUP_INDICES = CSV_DATASET_MAP.values()
+        .stream()
+        .filter(td -> td.settingFileName() != null && td.settingFileName().equals("lookup-settings.json"))
+        .map(CsvTestsDataLoader.TestDataset::indexName)
+        .collect(Collectors.toSet());
+
+    public static final Set<String> LOOKUP_ENDPOINTS = LOOKUP_INDICES.stream().map(i -> "/" + i + "/_bulk").collect(Collectors.toSet());
+
+    public static final Set<String> ENRICH_ENDPOINTS = ENRICH_SOURCE_INDICES.stream()
+        .map(i -> "/" + i + "/_bulk")
+        .collect(Collectors.toSet());
+
     /**
      * Creates a new mock client that dispatches every request to both the local and remote clusters, excluding _bulk and _query requests.
      * - '_bulk' requests are randomly sent to either the local or remote cluster to populate data. Some spec tests, such as AVG,
@@ -199,15 +242,17 @@ static RestClient twoClients(RestClient localClient, RestClient remoteClient) th
                 return localClient.performRequest(request);
             } else if (endpoint.endsWith("/_bulk") && METADATA_INDICES.stream().anyMatch(i -> endpoint.equals("/" + i + "/_bulk"))) {
                 return remoteClient.performRequest(request);
-            } else if (endpoint.endsWith("/_bulk") && ENRICH_SOURCE_INDICES.stream().noneMatch(i -> endpoint.equals("/" + i + "/_bulk"))) {
-                return bulkClient.performRequest(request);
-            } else {
-                Request[] clones = cloneRequests(request, 2);
-                Response resp1 = remoteClient.performRequest(clones[0]);
-                Response resp2 = localClient.performRequest(clones[1]);
-                assertEquals(resp1.getStatusLine().getStatusCode(), resp2.getStatusLine().getStatusCode());
-                return resp2;
-            }
+            } else if (endpoint.endsWith("/_bulk")
+                && ENRICH_ENDPOINTS.contains(endpoint) == false
+                && LOOKUP_ENDPOINTS.contains(endpoint) == false) {
+                    return bulkClient.performRequest(request);
+                } else {
+                    Request[] clones = cloneRequests(request, 2);
+                    Response resp1 = remoteClient.performRequest(clones[0]);
+                    Response resp2 = localClient.performRequest(clones[1]);
+                    assertEquals(resp1.getStatusLine().getStatusCode(), resp2.getStatusLine().getStatusCode());
+                    return resp2;
+                }
         });
         doAnswer(invocation -> {
             IOUtils.close(localClient, remoteClient);
@@ -251,37 +296,32 @@ static CsvSpecReader.CsvTestCase convertToRemoteIndices(CsvSpecReader.CsvTestCas
         String query = testCase.query;
         String[] commands = query.split("\\|");
         String first = commands[0].trim();
+        // If true, we're using *:index, otherwise we're using *:index,index
+        boolean onlyRemotes = canUseRemoteIndicesOnly() && randomBoolean();
         if (commands[0].toLowerCase(Locale.ROOT).startsWith("from")) {
             String[] parts = commands[0].split("(?i)metadata");
             assert parts.length >= 1 : parts;
             String fromStatement = parts[0];
             String[] localIndices = fromStatement.substring("FROM ".length()).split(",");
-            final String remoteIndices;
-            if (canUseRemoteIndicesOnly() && randomBoolean()) {
-                remoteIndices = Arrays.stream(localIndices)
-                    .map(index -> unquoteAndRequoteAsRemote(index.trim(), true))
-                    .collect(Collectors.joining(","));
-            } else {
-                remoteIndices = Arrays.stream(localIndices)
-                    .map(index -> unquoteAndRequoteAsRemote(index.trim(), false))
-                    .collect(Collectors.joining(","));
+            if (Arrays.stream(localIndices).anyMatch(i -> LOOKUP_INDICES.contains(i.trim().toLowerCase(Locale.ROOT)))) {
+                // If the query contains lookup indices, use only remotes to avoid duplication
+                onlyRemotes = true;
             }
+            final boolean onlyRemotesFinal = onlyRemotes;
+            final String remoteIndices = Arrays.stream(localIndices)
+                .map(index -> unquoteAndRequoteAsRemote(index.trim(), onlyRemotesFinal))
+                .collect(Collectors.joining(","));
             var newFrom = "FROM " + remoteIndices + " " + commands[0].substring(fromStatement.length());
             testCase.query = newFrom + query.substring(first.length());
         }
         if (commands[0].toLowerCase(Locale.ROOT).startsWith("ts ")) {
             String[] parts = commands[0].split("\\s+");
             assert parts.length >= 2 : commands[0];
             String[] indices = parts[1].split(",");
-            if (canUseRemoteIndicesOnly() && randomBoolean()) {
-                parts[1] = Arrays.stream(indices)
-                    .map(index -> unquoteAndRequoteAsRemote(index.trim(), true))
-                    .collect(Collectors.joining(","));
-            } else {
-                parts[1] = Arrays.stream(indices)
-                    .map(index -> unquoteAndRequoteAsRemote(index.trim(), false))
-                    .collect(Collectors.joining(","));
-            }
+            final boolean onlyRemotesFinal = onlyRemotes;
+            parts[1] = Arrays.stream(indices)
+                .map(index -> unquoteAndRequoteAsRemote(index.trim(), onlyRemotesFinal))
+                .collect(Collectors.joining(","));
             String newNewMetrics = String.join(" ", parts);
             testCase.query = newNewMetrics + query.substring(first.length());
         }
@@ -359,9 +399,7 @@ protected boolean supportsInferenceTestService() {
 
     @Override
     protected boolean supportsIndexModeLookup() throws IOException {
-        // CCS does not yet support JOIN_LOOKUP_V10 and clusters falsely report they have this capability
-        // return hasCapabilities(List.of(JOIN_LOOKUP_V10.capabilityName()));
-        return false;
+        return hasCapabilities(List.of(JOIN_LOOKUP_V12.capabilityName()));
     }
 
     @Override

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClustersIT.java

@@ -72,6 +72,9 @@ record Doc(int id, String color, long data) {
     List<Doc> localDocs = List.of();
     final String remoteIndex = "test-remote-index";
     List<Doc> remoteDocs = List.of();
+    final String lookupIndexLocal = "test-lookup-index-local";
+    final String lookupIndexRemote = "test-lookup-index-remote";
+    final String lookupAlias = "test-lookup-index";
 
     @Before
     public void setUpIndices() throws Exception {
@@ -109,6 +112,44 @@ public void setUpIndices() throws Exception {
         }
     }
 
+    private void setupLookupIndices() throws IOException {
+        RestClient localClient = client();
+        final String mapping = """
+             "properties": {
+               "data": { "type": "long" },
+               "morecolor": { "type": "keyword" }
+             }
+            """;
+        var lookupDocs = IntStream.range(0, between(1, 5))
+            .mapToObj(n -> new Doc(n, randomFrom("red", "yellow", "green"), randomIntBetween(1, 1000)))
+            .toList();
+        createIndex(
+            localClient,
+            lookupIndexLocal,
+            Settings.builder().put("index.number_of_shards", 1).put("index.mode", "lookup").build(),
+            mapping,
+            "\"" + lookupAlias + "\":{}"
+        );
+        indexDocs(localClient, lookupIndexLocal, lookupDocs);
+        try (RestClient remoteClient = remoteClusterClient()) {
+            createIndex(
+                remoteClient,
+                lookupIndexRemote,
+                Settings.builder().put("index.number_of_shards", 1).put("index.mode", "lookup").build(),
+                mapping,
+                "\"" + lookupAlias + "\":{}"
+            );
+            indexDocs(remoteClient, lookupIndexRemote, lookupDocs);
+        }
+    }
+
+    public void wipeLookupIndices() throws IOException {
+        try (RestClient remoteClient = remoteClusterClient()) {
+            deleteIndex(remoteClient, lookupIndexRemote);
+        }
+        deleteIndex(client(), lookupIndexLocal);
+    }
+
     @After
     public void wipeIndices() throws Exception {
         try (RestClient remoteClient = remoteClusterClient()) {
@@ -416,6 +457,64 @@ public void testStats() throws IOException {
         assertThat(clusterData, hasKey("took"));
     }
 
+    public void testLookupJoinAliases() throws IOException {
+        assumeTrue(
+            "Local cluster does not support multiple LOOKUP JOIN aliases",
+            supportsLookupJoinAliases(Clusters.localClusterVersion())
+        );
+        assumeTrue(
+            "Remote cluster does not support multiple LOOKUP JOIN aliases",
+            supportsLookupJoinAliases(Clusters.remoteClusterVersion())
+        );
+        try {
+            setupLookupIndices();
+            Map<String, Object> result = run(
+                "FROM test-local-index,*:test-remote-index | LOOKUP JOIN test-lookup-index ON data | STATS c = COUNT(*)",
+                true
+            );
+            var columns = List.of(Map.of("name", "c", "type", "long"));
+            var values = List.of(List.of(localDocs.size() + remoteDocs.size()));
+            assertResultMap(true, result, columns, values, false);
+        } finally {
+            wipeLookupIndices();
+        }
+    }
+
+    public void testLookupJoinAliasesSkipOld() throws IOException {
+        assumeTrue(
+            "Local cluster does not support multiple LOOKUP JOIN aliases",
+            supportsLookupJoinAliases(Clusters.localClusterVersion())
+        );
+        assumeFalse(
+            "Remote cluster should not support multiple LOOKUP JOIN aliases",
+            supportsLookupJoinAliases(Clusters.remoteClusterVersion())
+        );
+        try {
+            setupLookupIndices();
+            Map<String, Object> result = run(
+                "FROM test-local-index,*:test-remote-index | LOOKUP JOIN test-lookup-index ON data | STATS c = COUNT(*)",
+                true
+            );
+            var columns = List.of(Map.of("name", "c", "type", "long"));
+            var values = List.of(List.of(localDocs.size()));
+
+            MapMatcher mapMatcher = getResultMatcher(true, false, result.containsKey("documents_found")).extraOk();
+            mapMatcher = mapMatcher.entry("_clusters", any(Map.class));
+            mapMatcher = mapMatcher.entry("is_partial", true);
+            assertMap(result, mapMatcher.entry("columns", columns).entry("values", values));
+            // check that the remote is skipped
+            @SuppressWarnings("unchecked")
+            Map<String, Object> clusters = (Map<String, Object>) result.get("_clusters");
+            @SuppressWarnings("unchecked")
+            Map<String, Object> details = (Map<String, Object>) clusters.get("details");
+            @SuppressWarnings("unchecked")
+            Map<String, Object> remoteCluster = (Map<String, Object>) details.get("remote_cluster");
+            assertThat(remoteCluster.get("status"), equalTo("skipped"));
+        } finally {
+            wipeLookupIndices();
+        }
+    }
+
     public void testLikeIndex() throws Exception {
 
         boolean includeCCSMetadata = includeCCSMetadata();
@@ -576,6 +675,10 @@ private static boolean capabilitiesEndpointAvailable() {
         return Clusters.localClusterVersion().onOrAfter(Version.V_8_15_0);
     }
 
+    private static boolean supportsLookupJoinAliases(Version version) {
+        return version.onOrAfter(Version.V_9_2_0);
+    }
+
     private static boolean includeCCSMetadata() {
         return ccsMetadataAvailable() && randomBoolean();
     }

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/RequestIndexFilteringIT.java

@@ -18,6 +18,7 @@
 import org.elasticsearch.test.MapMatcher;
 import org.elasticsearch.test.TestClustersThreadFilter;
 import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.xpack.esql.action.EsqlCapabilities;
 import org.elasticsearch.xpack.esql.qa.rest.RequestIndexFilteringTestCase;
 import org.elasticsearch.xpack.esql.qa.rest.RestEsqlTestCase;
 import org.hamcrest.Matcher;
@@ -35,6 +36,8 @@
 import static org.elasticsearch.test.ListMatcher.matchesList;
 import static org.elasticsearch.test.MapMatcher.assertMap;
 import static org.elasticsearch.test.MapMatcher.matchesMap;
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
@@ -153,6 +156,35 @@ private static boolean checkVersion(org.elasticsearch.Version version) {
             || (version.onOrAfter(Version.fromString("8.19.0")) && version.before(Version.fromString("9.0.0")));
     }
 
+    public void testIndicesDontExistWithRemoteLookupJoin() throws IOException {
+        assumeTrue("Only works with remote LOOKUP JOIN support", EsqlCapabilities.Cap.ENABLE_LOOKUP_JOIN_ON_REMOTE.isEnabled());
+        // This check is for "local" cluster - which is different from test runner actually, so it could be old
+        assumeTrue(
+            "Only works with remote LOOKUP JOIN support",
+            clusterHasCapability(
+                client(),
+                "POST",
+                "_query",
+                List.of(),
+                List.of(EsqlCapabilities.Cap.ENABLE_LOOKUP_JOIN_ON_REMOTE.capabilityName())
+            ).orElse(false)
+        );
+
+        int docsTest1 = randomIntBetween(1, 5);
+        indexTimestampData(docsTest1, "test1", "2024-11-26", "id1");
+
+        var pattern = "FROM test1,*:test1";
+        ResponseException e = expectThrows(
+            ResponseException.class,
+            () -> runEsql(timestampFilter("gte", "2020-01-01").query(pattern + " | LOOKUP JOIN foo ON id1"))
+        );
+        assertEquals(400, e.getResponse().getStatusLine().getStatusCode());
+        assertThat(
+            e.getMessage(),
+            allOf(containsString("verification_exception"), containsString("Unknown index [foo,remote_cluster:foo]"))
+        );
+    }
+
     // We need a separate test since remote missing indices and local missing indices now work differently
     public void testIndicesDontExistRemote() throws IOException {
         // Exclude old versions