[8.19] ES|QL: Add support for LOOKUP JOIN on aliases (#128519) (#128833)

* ES|QL: Add support for LOOKUP JOIN on aliases (#128519)

* Fix test

docs/changelog/128519.yaml

@@ -0,0 +1,5 @@
+pr: 128519
+summary: Add support for LOOKUP JOIN on aliases
+area: ES|QL
+type: enhancement
+issues: []

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

@@ -231,6 +231,7 @@ public class TransportVersions {
     public static final TransportVersion INFERENCE_CUSTOM_SERVICE_ADDED_8_19 = def(8_841_0_39);
     public static final TransportVersion IDP_CUSTOM_SAML_ATTRIBUTES_ADDED_8_19 = def(8_841_0_40);
     public static final TransportVersion DATA_STREAM_OPTIONS_API_REMOVE_INCLUDE_DEFAULTS_8_19 = def(8_841_0_41);
+    public static final TransportVersion JOIN_ON_ALIASES_8_19 = def(8_841_0_42);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

x-pack/plugin/build.gradle

@@ -227,6 +227,7 @@ tasks.named("yamlRestTestV7CompatTransform").configure({ task ->
   task.skipTest("esql/40_unsupported_types/unsupported", "TODO: support for subset of metric fields")
   task.skipTest("esql/40_unsupported_types/unsupported with sort", "TODO: support for subset of metric fields")
   task.skipTest("esql/63_enrich_int_range/Invalid age as double", "TODO: require disable allow_partial_results")
+  task.skipTest("esql/191_lookup_join_on_datastreams/data streams not supported in LOOKUP JOIN", "Added support for aliases in JOINs")
 })
 
 

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/lookup/QueryList.java

@@ -34,8 +34,10 @@ import org.elasticsearch.index.mapper.GeoShapeQueryable;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.RangeFieldMapper;
 import org.elasticsearch.index.query.SearchExecutionContext;
+import org.elasticsearch.search.internal.AliasFilter;
 
 import java.io.IOException;
+import java.io.UncheckedIOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.IntFunction;
@@ -45,12 +47,20 @@ import java.util.function.IntFunction;
  */
 public abstract class QueryList {
     protected final SearchExecutionContext searchExecutionContext;
+    protected final AliasFilter aliasFilter;
     protected final MappedFieldType field;
     protected final Block block;
     protected final boolean onlySingleValues;
 
-    protected QueryList(MappedFieldType field, SearchExecutionContext searchExecutionContext, Block block, boolean onlySingleValues) {
+    protected QueryList(
+        MappedFieldType field,
+        SearchExecutionContext searchExecutionContext,
+        AliasFilter aliasFilter,
+        Block block,
+        boolean onlySingleValues
+    ) {
         this.searchExecutionContext = searchExecutionContext;
+        this.aliasFilter = aliasFilter;
         this.field = field;
         this.block = block;
         this.onlySingleValues = onlySingleValues;
@@ -78,6 +88,17 @@ public abstract class QueryList {
 
         Query query = doGetQuery(position, firstValueIndex, valueCount);
 
+        if (aliasFilter != null && aliasFilter != AliasFilter.EMPTY) {
+            BooleanQuery.Builder builder = new BooleanQuery.Builder();
+            builder.add(query, BooleanClause.Occur.FILTER);
+            try {
+                builder.add(aliasFilter.getQueryBuilder().toQuery(searchExecutionContext), BooleanClause.Occur.FILTER);
+                query = builder.build();
+            } catch (IOException e) {
+                throw new UncheckedIOException("Error while building query for alias filter", e);
+            }
+        }
+
         if (onlySingleValues) {
             query = wrapSingleValueQuery(query);
         }
@@ -121,7 +142,12 @@ public abstract class QueryList {
      * using only the {@link ElementType} of the {@link Block} to determine the
      * query.
      */
-    public static QueryList rawTermQueryList(MappedFieldType field, SearchExecutionContext searchExecutionContext, Block block) {
+    public static QueryList rawTermQueryList(
+        MappedFieldType field,
+        SearchExecutionContext searchExecutionContext,
+        AliasFilter aliasFilter,
+        Block block
+    ) {
         IntFunction<Object> blockToJavaObject = switch (block.elementType()) {
             case BOOLEAN -> {
                 BooleanBlock booleanBlock = (BooleanBlock) block;
@@ -153,17 +179,22 @@ public abstract class QueryList {
             case AGGREGATE_METRIC_DOUBLE -> throw new IllegalArgumentException("can't read values from [aggregate metric double] block");
             case UNKNOWN -> throw new IllegalArgumentException("can't read values from [" + block + "]");
         };
-        return new TermQueryList(field, searchExecutionContext, block, false, blockToJavaObject);
+        return new TermQueryList(field, searchExecutionContext, aliasFilter, block, false, blockToJavaObject);
     }
 
     /**
      * Returns a list of term queries for the given field and the input block of
      * {@code ip} field values.
      */
-    public static QueryList ipTermQueryList(MappedFieldType field, SearchExecutionContext searchExecutionContext, BytesRefBlock block) {
+    public static QueryList ipTermQueryList(
+        MappedFieldType field,
+        SearchExecutionContext searchExecutionContext,
+        AliasFilter aliasFilter,
+        BytesRefBlock block
+    ) {
         BytesRef scratch = new BytesRef();
         byte[] ipBytes = new byte[InetAddressPoint.BYTES];
-        return new TermQueryList(field, searchExecutionContext, block, false, offset -> {
+        return new TermQueryList(field, searchExecutionContext, aliasFilter, block, false, offset -> {
             final var bytes = block.getBytesRef(offset, scratch);
             if (ipBytes.length != bytes.length) {
                 // Lucene only support 16-byte IP addresses, even IPv4 is encoded in 16 bytes
@@ -178,10 +209,16 @@ public abstract class QueryList {
      * Returns a list of term queries for the given field and the input block of
      * {@code date} field values.
      */
-    public static QueryList dateTermQueryList(MappedFieldType field, SearchExecutionContext searchExecutionContext, LongBlock block) {
+    public static QueryList dateTermQueryList(
+        MappedFieldType field,
+        SearchExecutionContext searchExecutionContext,
+        AliasFilter aliasFilter,
+        LongBlock block
+    ) {
         return new TermQueryList(
             field,
             searchExecutionContext,
+            aliasFilter,
             block,
             false,
             field instanceof RangeFieldMapper.RangeFieldType rangeFieldType
@@ -193,8 +230,14 @@ public abstract class QueryList {
     /**
      * Returns a list of geo_shape queries for the given field and the input block.
      */
-    public static QueryList geoShapeQueryList(MappedFieldType field, SearchExecutionContext searchExecutionContext, Block block) {
-        return new GeoShapeQueryList(field, searchExecutionContext, block, false);
+
+    public static QueryList geoShapeQueryList(
+        MappedFieldType field,
+        SearchExecutionContext searchExecutionContext,
+        AliasFilter aliasFilter,
+        Block block
+    ) {
+        return new GeoShapeQueryList(field, searchExecutionContext, aliasFilter, block, false);
     }
 
     private static class TermQueryList extends QueryList {
@@ -203,17 +246,18 @@ public abstract class QueryList {
         private TermQueryList(
             MappedFieldType field,
             SearchExecutionContext searchExecutionContext,
+            AliasFilter aliasFilter,
             Block block,
             boolean onlySingleValues,
             IntFunction<Object> blockValueReader
         ) {
-            super(field, searchExecutionContext, block, onlySingleValues);
+            super(field, searchExecutionContext, aliasFilter, block, onlySingleValues);
             this.blockValueReader = blockValueReader;
         }
 
         @Override
         public TermQueryList onlySingleValues() {
-            return new TermQueryList(field, searchExecutionContext, block, true, blockValueReader);
+            return new TermQueryList(field, searchExecutionContext, aliasFilter, block, true, blockValueReader);
         }
 
         @Override
@@ -241,10 +285,11 @@ public abstract class QueryList {
         private GeoShapeQueryList(
             MappedFieldType field,
             SearchExecutionContext searchExecutionContext,
+            AliasFilter aliasFilter,
             Block block,
             boolean onlySingleValues
         ) {
-            super(field, searchExecutionContext, block, onlySingleValues);
+            super(field, searchExecutionContext, aliasFilter, block, onlySingleValues);
 
             this.blockValueReader = blockToGeometry(block);
             this.shapeQuery = shapeQuery();
@@ -252,7 +297,7 @@ public abstract class QueryList {
 
         @Override
         public GeoShapeQueryList onlySingleValues() {
-            return new GeoShapeQueryList(field, searchExecutionContext, block, true);
+            return new GeoShapeQueryList(field, searchExecutionContext, aliasFilter, block, true);
         }
 
         @Override

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/lookup/EnrichQuerySourceOperatorTests.java

@@ -41,6 +41,7 @@ import org.elasticsearch.index.mapper.KeywordFieldMapper;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
+import org.elasticsearch.search.internal.AliasFilter;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.After;
 import org.junit.Before;
@@ -83,7 +84,7 @@ public class EnrichQuerySourceOperatorTests extends ESTestCase {
             var inputTerms = makeTermsBlock(List.of(List.of("b2"), List.of("c1", "a2"), List.of("z2"), List.of(), List.of("a3"), List.of()))
         ) {
             MappedFieldType uidField = new KeywordFieldMapper.KeywordFieldType("uid");
-            QueryList queryList = QueryList.rawTermQueryList(uidField, directoryData.searchExecutionContext, inputTerms);
+            QueryList queryList = QueryList.rawTermQueryList(uidField, directoryData.searchExecutionContext, AliasFilter.EMPTY, inputTerms);
             assertThat(queryList.getPositionCount(), equalTo(6));
             assertThat(queryList.getQuery(0), equalTo(new TermQuery(new Term("uid", new BytesRef("b2")))));
             assertThat(queryList.getQuery(1), equalTo(new TermInSetQuery("uid", List.of(new BytesRef("c1"), new BytesRef("a2")))));
@@ -154,7 +155,12 @@ public class EnrichQuerySourceOperatorTests extends ESTestCase {
         }).toList();
 
         try (var directoryData = makeDirectoryWith(directoryTermsList); var inputTerms = makeTermsBlock(inputTermsList)) {
-            var queryList = QueryList.rawTermQueryList(directoryData.field, directoryData.searchExecutionContext, inputTerms);
+            var queryList = QueryList.rawTermQueryList(
+                directoryData.field,
+                directoryData.searchExecutionContext,
+                AliasFilter.EMPTY,
+                inputTerms
+            );
             int maxPageSize = between(1, 256);
             var warnings = Warnings.createWarnings(DriverContext.WarningsMode.IGNORE, 0, 0, "test enrich");
             EnrichQuerySourceOperator queryOperator = new EnrichQuerySourceOperator(
@@ -192,8 +198,12 @@ public class EnrichQuerySourceOperatorTests extends ESTestCase {
                 List.of(List.of("b2"), List.of("c1", "a2"), List.of("z2"), List.of(), List.of("a3"), List.of("a3", "a2", "z2", "xx"))
             )
         ) {
-            QueryList queryList = QueryList.rawTermQueryList(directoryData.field, directoryData.searchExecutionContext, inputTerms)
-                .onlySingleValues();
+            QueryList queryList = QueryList.rawTermQueryList(
+                directoryData.field,
+                directoryData.searchExecutionContext,
+                AliasFilter.EMPTY,
+                inputTerms
+            ).onlySingleValues();
             // pos -> terms -> docs
             // -----------------------------
             // 0 -> [b2] -> []

x-pack/plugin/esql/qa/security/src/javaRestTest/java/org/elasticsearch/xpack/esql/EsqlSecurityIT.java

@@ -42,6 +42,8 @@ import static org.elasticsearch.test.MapMatcher.assertMap;
 import static org.elasticsearch.test.MapMatcher.matchesMap;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.nullValue;
 
 public class EsqlSecurityIT extends ESRestTestCase {
     @ClassRule
@@ -58,10 +60,14 @@ public class EsqlSecurityIT extends ESRestTestCase {
         .user("user5", "x-pack-test-password", "user5", false)
         .user("fls_user", "x-pack-test-password", "fls_user", false)
         .user("fls_user2", "x-pack-test-password", "fls_user2", false)
+        .user("fls_user2_alias", "x-pack-test-password", "fls_user2_alias", false)
         .user("fls_user3", "x-pack-test-password", "fls_user3", false)
+        .user("fls_user3_alias", "x-pack-test-password", "fls_user3_alias", false)
         .user("fls_user4_1", "x-pack-test-password", "fls_user4_1", false)
+        .user("fls_user4_1_alias", "x-pack-test-password", "fls_user4_1_alias", false)
         .user("dls_user", "x-pack-test-password", "dls_user", false)
         .user("metadata1_read2", "x-pack-test-password", "metadata1_read2", false)
+        .user("metadata1_alias_read2", "x-pack-test-password", "metadata1_alias_read2", false)
         .user("alias_user1", "x-pack-test-password", "alias_user1", false)
         .user("alias_user2", "x-pack-test-password", "alias_user2", false)
         .user("logs_foo_all", "x-pack-test-password", "logs_foo_all", false)
@@ -156,6 +162,12 @@ public class EsqlSecurityIT extends ESRestTestCase {
                             }
                           }
                         },
+                        {
+                          "add": {
+                            "alias": "lookup-second-alias",
+                            "index": "lookup-user2"
+                          }
+                        },
                         {
                           "add": {
                             "alias": "second-alias",
@@ -193,6 +205,17 @@ public class EsqlSecurityIT extends ESRestTestCase {
             }
             """);
         assertOK(client().performRequest(request));
+
+        request = new Request("POST", "_security/user/fls_user4_alias");
+        request.setJsonEntity("""
+            {
+              "password" : "x-pack-test-password",
+              "roles" : [ "fls_user4_1_alias", "fls_user4_2_alias" ],
+              "full_name" : "Test Role",
+              "email" : "test.role@example.com"
+            }
+            """);
+        assertOK(client().performRequest(request));
     }
 
     protected MapMatcher responseMatcher(Map<String, Object> result) {
@@ -583,22 +606,76 @@ public class EsqlSecurityIT extends ESRestTestCase {
         );
         assertThat(respMap.get("values"), equalTo(List.of(List.of(40.0, "sales"))));
 
-        // Aliases are not allowed in LOOKUP JOIN
-        var resp2 = expectThrows(
+        // user is not allowed to use the alias (but is allowed to use the index)
+        expectThrows(
             ResponseException.class,
-            () -> runESQLCommand("alias_user1", "ROW x = 31.0 | EVAL value = x | LOOKUP JOIN lookup-first-alias ON value | KEEP x, org")
+            () -> runESQLCommand(
+                "metadata1_read2",
+                "ROW x = 40.0 | EVAL value = x | LOOKUP JOIN lookup-second-alias ON value | KEEP x, org"
+            )
         );
 
-        assertThat(resp2.getMessage(), containsString("Aliases and index patterns are not allowed for LOOKUP JOIN [lookup-first-alias]"));
-        assertThat(resp2.getResponse().getStatusLine().getStatusCode(), equalTo(HttpStatus.SC_BAD_REQUEST));
-
-        // Aliases are not allowed in LOOKUP JOIN, regardless of alias filters
-        resp2 = expectThrows(
+        // user is not allowed to use the index (but is allowed to use the alias)
+        expectThrows(
             ResponseException.class,
-            () -> runESQLCommand("alias_user1", "ROW x = 123.0 | EVAL value = x | LOOKUP JOIN lookup-first-alias ON value | KEEP x, org")
+            () -> runESQLCommand("metadata1_alias_read2", "ROW x = 40.0 | EVAL value = x | LOOKUP JOIN lookup-user2 ON value | KEEP x, org")
+        );
+
+        // user has permission on the alias, and can read the key
+        resp = runESQLCommand(
+            "metadata1_alias_read2",
+            "ROW x = 40.0 | EVAL value = x | LOOKUP JOIN lookup-second-alias ON value | KEEP x, org"
+        );
+        assertOK(resp);
+        respMap = entityAsMap(resp);
+        assertThat(
+            respMap.get("columns"),
+            equalTo(List.of(Map.of("name", "x", "type", "double"), Map.of("name", "org", "type", "keyword")))
         );
-        assertThat(resp2.getMessage(), containsString("Aliases and index patterns are not allowed for LOOKUP JOIN [lookup-first-alias]"));
-        assertThat(resp2.getResponse().getStatusLine().getStatusCode(), equalTo(HttpStatus.SC_BAD_REQUEST));
+        assertThat(respMap.get("values"), equalTo(List.of(List.of(40.0, "sales"))));
+
+        // user has permission on the alias, but can't read the key (doc level security at role level)
+        resp = runESQLCommand(
+            "metadata1_alias_read2",
+            "ROW x = 32.0 | EVAL value = x | LOOKUP JOIN lookup-second-alias ON value | KEEP x, org"
+        );
+        assertOK(resp);
+        respMap = entityAsMap(resp);
+        assertThat(
+            respMap.get("columns"),
+            equalTo(List.of(Map.of("name", "x", "type", "double"), Map.of("name", "org", "type", "keyword")))
+        );
+        List<?> values = (List<?>) respMap.get("values");
+        assertThat(values.size(), is(1));
+        List<?> row = (List<?>) values.get(0);
+        assertThat(row.size(), is(2));
+        assertThat(row.get(0), is(32.0));
+        assertThat(row.get(1), is(nullValue()));
+
+        // user has permission on the alias, the alias has a filter that doesn't allow to see the value
+        resp = runESQLCommand("alias_user1", "ROW x = 12.0 | EVAL value = x | LOOKUP JOIN lookup-first-alias ON value | KEEP x, org");
+        assertOK(resp);
+        respMap = entityAsMap(resp);
+        assertThat(
+            respMap.get("columns"),
+            equalTo(List.of(Map.of("name", "x", "type", "double"), Map.of("name", "org", "type", "keyword")))
+        );
+        values = (List<?>) respMap.get("values");
+        assertThat(values.size(), is(1));
+        row = (List<?>) values.get(0);
+        assertThat(row.size(), is(2));
+        assertThat(row.get(0), is(12.0));
+        assertThat(row.get(1), is(nullValue()));
+
+        // user has permission on the alias, the alias has a filter that allows to see the value
+        resp = runESQLCommand("alias_user1", "ROW x = 31.0 | EVAL value = x | LOOKUP JOIN lookup-first-alias ON value | KEEP x, org");
+        assertOK(resp);
+        respMap = entityAsMap(resp);
+        assertThat(
+            respMap.get("columns"),
+            equalTo(List.of(Map.of("name", "x", "type", "double"), Map.of("name", "org", "type", "keyword")))
+        );
+        assertThat(respMap.get("values"), equalTo(List.of(List.of(31.0, "sales"))));
     }
 
     @SuppressWarnings("unchecked")
@@ -709,6 +786,64 @@ public class EsqlSecurityIT extends ESRestTestCase {
         assertThat(error.getResponse().getStatusLine().getStatusCode(), equalTo(HttpStatus.SC_BAD_REQUEST));
     }
 
+    public void testLookupJoinFieldLevelSecurityOnAlias() throws Exception {
+        assumeTrue(
+            "Requires LOOKUP JOIN capability",
+            EsqlSpecTestCase.hasCapabilities(adminClient(), List.of(EsqlCapabilities.Cap.JOIN_LOOKUP_V12.capabilityName()))
+        );
+
+        Response resp = runESQLCommand("fls_user2_alias", "ROW x = 40.0 | EVAL value = x | LOOKUP JOIN lookup-second-alias ON value");
+        assertOK(resp);
+        Map<String, Object> respMap = entityAsMap(resp);
+        assertThat(
+            respMap.get("columns"),
+            equalTo(
+                List.of(
+                    Map.of("name", "x", "type", "double"),
+                    Map.of("name", "value", "type", "double"),
+                    Map.of("name", "org", "type", "keyword")
+                )
+            )
+        );
+
+        resp = runESQLCommand("fls_user3_alias", "ROW x = 40.0 | EVAL value = x | LOOKUP JOIN lookup-second-alias ON value");
+        assertOK(resp);
+        respMap = entityAsMap(resp);
+        assertThat(
+            respMap.get("columns"),
+            equalTo(
+                List.of(
+                    Map.of("name", "x", "type", "double"),
+                    Map.of("name", "value", "type", "double"),
+                    Map.of("name", "org", "type", "keyword"),
+                    Map.of("name", "other", "type", "keyword")
+                )
+            )
+
+        );
+
+        resp = runESQLCommand("fls_user4_alias", "ROW x = 40.0 | EVAL value = x | LOOKUP JOIN lookup-second-alias ON value");
+        assertOK(resp);
+        respMap = entityAsMap(resp);
+        assertThat(
+            respMap.get("columns"),
+            equalTo(
+                List.of(
+                    Map.of("name", "x", "type", "double"),
+                    Map.of("name", "value", "type", "double"),
+                    Map.of("name", "org", "type", "keyword")
+                )
+            )
+        );
+
+        ResponseException error = expectThrows(
+            ResponseException.class,
+            () -> runESQLCommand("fls_user4_1_alias", "ROW x = 40.0 | EVAL value = x | LOOKUP JOIN lookup-second-alias ON value")
+        );
+        assertThat(error.getMessage(), containsString("Unknown column [value] in right side of join"));
+        assertThat(error.getResponse().getStatusLine().getStatusCode(), equalTo(HttpStatus.SC_BAD_REQUEST));
+    }
+
     public void testLookupJoinIndexForbidden() throws Exception {
         assumeTrue(
             "Requires LOOKUP JOIN capability",

x-pack/plugin/esql/qa/security/src/javaRestTest/resources/roles.yml

@@ -40,6 +40,17 @@ metadata1_read2:
     - names: [ 'index-user2', 'lookup-user2' ]
       privileges: [ 'read' ]
 
+metadata1_alias_read2:
+  cluster: []
+  indices:
+    - names: [ 'index-user1', 'lookup-first-alias' ]
+      privileges: [ 'view_index_metadata' ]
+    - names: [ 'index-user2' ]
+      privileges: [ 'read' ]
+    - names: [ 'lookup-second-alias' ]
+      privileges: [ 'read' ]
+      query: '{"match": {"org": "sales"}}'
+
 alias_user1:
   cluster: []
   indices:
@@ -101,6 +112,14 @@ fls_user2:
       field_security:
         grant: [ "org", "value" ]
 
+fls_user2_alias:
+  cluster: []
+  indices:
+    - names: [ 'lookup-second-alias' ]
+      privileges: [ 'read' ]
+      field_security:
+        grant: [ "org", "value" ]
+
 fls_user3:
   cluster: []
   indices:
@@ -109,6 +128,15 @@ fls_user3:
       field_security:
         grant: [ "org", "value", "other" ]
 
+fls_user3_alias:
+  cluster: []
+  indices:
+    - names: [ 'lookup-second-alias' ]
+      privileges: [ 'read' ]
+      field_security:
+        grant: [ "org", "value", "other" ]
+
+
 fls_user4_1:
   cluster: []
   indices:
@@ -117,6 +145,14 @@ fls_user4_1:
       field_security:
         grant: [ "org" ]
 
+fls_user4_1_alias:
+  cluster: []
+  indices:
+    - names: [ 'lookup-second-alias' ]
+      privileges: [ 'read' ]
+      field_security:
+        grant: [ "org" ]
+
 fls_user4_2:
   cluster: []
   indices:
@@ -125,6 +161,14 @@ fls_user4_2:
       field_security:
         grant: [ "value" ]
 
+fls_user4_2_alias:
+  cluster: []
+  indices:
+    - names: [ 'lookup-second-alias' ]
+      privileges: [ 'read' ]
+      field_security:
+        grant: [ "value" ]
+
 dls_user:
   cluster: []
   indices:

x-pack/plugin/esql/qa/server/mixed-cluster/build.gradle

@@ -12,7 +12,7 @@ apply plugin: 'elasticsearch.bwc-test'
 
 restResources {
   restApi {
-    include '_common', 'bulk', 'get', 'indices', 'esql', 'xpack', 'enrich', 'cluster', 'capabilities'
+    include '_common', 'bulk', 'get', 'indices', 'esql', 'xpack', 'enrich', 'cluster', 'capabilities', 'index'
   }
   restTests {
     includeXpack 'esql'

x-pack/plugin/esql/qa/server/multi-node/build.gradle

@@ -22,7 +22,7 @@ tasks.named('javaRestTest') {
 
 restResources {
   restApi {
-    include '_common', 'bulk', 'get', 'indices', 'esql', 'xpack', 'enrich', 'cluster', 'capabilities'
+    include '_common', 'bulk', 'get', 'indices', 'esql', 'xpack', 'enrich', 'cluster', 'capabilities', 'index'
   }
   restTests {
     includeXpack 'esql'

x-pack/plugin/esql/qa/server/single-node/build.gradle

@@ -28,7 +28,7 @@ dependencies {
 
 restResources {
   restApi {
-    include '_common', 'bulk', 'get', 'indices', 'esql', 'xpack', 'enrich', 'cluster', 'capabilities'
+    include '_common', 'bulk', 'get', 'indices', 'esql', 'xpack', 'enrich', 'cluster', 'capabilities', 'index'
   }
   restTests {
     includeXpack 'esql'

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

@@ -220,6 +220,7 @@ public class LookupFromIndexIT extends AbstractEsqlIntegTestCase {
                 ctx -> internalCluster().getInstance(TransportEsqlQueryAction.class, finalNodeWithShard).getLookupFromIndexService(),
                 keyType,
                 "lookup",
+                "lookup",
                 "key",
                 List.of(new Alias(Source.EMPTY, "l", new ReferenceAttribute(Source.EMPTY, "l", DataType.LONG))),
                 Source.EMPTY

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -899,7 +899,12 @@ public class EsqlCapabilities {
         /**
          * Support parameters for LiMIT command.
          */
-        PARAMETER_FOR_LIMIT;
+        PARAMETER_FOR_LIMIT,
+
+        /**
+         * Enable support for index aliases in lookup joins
+         */
+        ENABLE_LOOKUP_JOIN_ON_ALIASES(JOIN_LOOKUP_V12.isEnabled());
 
         private final boolean enabled;
 

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

@@ -14,6 +14,7 @@ import org.elasticsearch.action.UnavailableShardsException;
 import org.elasticsearch.action.support.ChannelActionListener;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.ShardIterator;
 import org.elasticsearch.cluster.routing.ShardRouting;
@@ -52,6 +53,7 @@ import org.elasticsearch.index.mapper.BlockLoader;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.internal.AliasFilter;
 import org.elasticsearch.search.internal.SearchContext;
@@ -121,8 +123,10 @@ import java.util.stream.IntStream;
 public abstract class AbstractLookupService<R extends AbstractLookupService.Request, T extends AbstractLookupService.TransportRequest> {
     private final String actionName;
     protected final ClusterService clusterService;
+    protected final IndicesService indicesService;
     private final LookupShardContextFactory lookupShardContextFactory;
     protected final TransportService transportService;
+    IndexNameExpressionResolver indexNameExpressionResolver;
     protected final Executor executor;
     private final BigArrays bigArrays;
     private final BlockFactory blockFactory;
@@ -140,8 +144,10 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
     AbstractLookupService(
         String actionName,
         ClusterService clusterService,
+        IndicesService indicesService,
         LookupShardContextFactory lookupShardContextFactory,
         TransportService transportService,
+        IndexNameExpressionResolver indexNameExpressionResolver,
         BigArrays bigArrays,
         BlockFactory blockFactory,
         boolean mergePages,
@@ -149,8 +155,10 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
     ) {
         this.actionName = actionName;
         this.clusterService = clusterService;
+        this.indicesService = indicesService;
         this.lookupShardContextFactory = lookupShardContextFactory;
         this.transportService = transportService;
+        this.indexNameExpressionResolver = indexNameExpressionResolver;
         this.executor = transportService.getThreadPool().executor(ThreadPool.Names.SEARCH);
         this.bigArrays = bigArrays;
         this.blockFactory = blockFactory;
@@ -177,7 +185,13 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
     /**
      * Build a list of queries to perform inside the actual lookup.
      */
-    protected abstract QueryList queryList(T request, SearchExecutionContext context, Block inputBlock, @Nullable DataType inputDataType);
+    protected abstract QueryList queryList(
+        T request,
+        SearchExecutionContext context,
+        AliasFilter aliasFilter,
+        Block inputBlock,
+        @Nullable DataType inputDataType
+    );
 
     /**
      * Build the response.
@@ -192,16 +206,17 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
     protected static QueryList termQueryList(
         MappedFieldType field,
         SearchExecutionContext searchExecutionContext,
+        AliasFilter aliasFilter,
         Block block,
         @Nullable DataType inputDataType
     ) {
         if (inputDataType == null) {
-            return QueryList.rawTermQueryList(field, searchExecutionContext, block);
+            return QueryList.rawTermQueryList(field, searchExecutionContext, aliasFilter, block);
         }
         return switch (inputDataType) {
-            case IP -> QueryList.ipTermQueryList(field, searchExecutionContext, (BytesRefBlock) block);
-            case DATETIME -> QueryList.dateTermQueryList(field, searchExecutionContext, (LongBlock) block);
-            default -> QueryList.rawTermQueryList(field, searchExecutionContext, block);
+            case IP -> QueryList.ipTermQueryList(field, searchExecutionContext, aliasFilter, (BytesRefBlock) block);
+            case DATETIME -> QueryList.dateTermQueryList(field, searchExecutionContext, aliasFilter, (LongBlock) block);
+            default -> QueryList.rawTermQueryList(field, searchExecutionContext, aliasFilter, block);
         };
     }
 
@@ -261,6 +276,14 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
         final List<Releasable> releasables = new ArrayList<>(6);
         boolean started = false;
         try {
+
+            var clusterState = clusterService.state();
+            AliasFilter aliasFilter = indicesService.buildAliasFilter(
+                clusterState,
+                request.shardId.getIndex().getName(),
+                indexNameExpressionResolver.resolveExpressions(clusterState, request.indexPattern)
+            );
+
             LookupShardContext shardContext = lookupShardContextFactory.create(request.shardId);
             releasables.add(shardContext.release);
             final LocalCircuitBreaker localBreaker = new LocalCircuitBreaker(
@@ -300,13 +323,15 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
                 }
             }
             releasables.add(finishPages);
-            QueryList queryList = queryList(request, shardContext.executionContext, inputBlock, request.inputDataType);
+
             var warnings = Warnings.createWarnings(
                 DriverContext.WarningsMode.COLLECT,
                 request.source.source().getLineNumber(),
                 request.source.source().getColumnNumber(),
                 request.source.text()
             );
+            QueryList queryList = queryList(request, shardContext.executionContext, aliasFilter, inputBlock, request.inputDataType);
+
             var queryOperator = new EnrichQuerySourceOperator(
                 driverContext.blockFactory(),
                 EnrichQuerySourceOperator.DEFAULT_MAX_PAGE_SIZE,
@@ -458,6 +483,7 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
     abstract static class Request {
         final String sessionId;
         final String index;
+        final String indexPattern;
         final DataType inputDataType;
         final Page inputPage;
         final List<NamedExpression> extractFields;
@@ -466,6 +492,7 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
         Request(
             String sessionId,
             String index,
+            String indexPattern,
             DataType inputDataType,
             Page inputPage,
             List<NamedExpression> extractFields,
@@ -473,6 +500,7 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
         ) {
             this.sessionId = sessionId;
             this.index = index;
+            this.indexPattern = indexPattern;
             this.inputDataType = inputDataType;
             this.inputPage = inputPage;
             this.extractFields = extractFields;
@@ -483,6 +511,7 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
     abstract static class TransportRequest extends org.elasticsearch.transport.TransportRequest implements IndicesRequest {
         final String sessionId;
         final ShardId shardId;
+        final String indexPattern;
         /**
          * For mixed clusters with nodes &lt;8.14, this will be null.
          */
@@ -498,6 +527,7 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
         TransportRequest(
             String sessionId,
             ShardId shardId,
+            String indexPattern,
             DataType inputDataType,
             Page inputPage,
             Page toRelease,
@@ -506,6 +536,7 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
         ) {
             this.sessionId = sessionId;
             this.shardId = shardId;
+            this.indexPattern = indexPattern;
             this.inputDataType = inputDataType;
             this.inputPage = inputPage;
             this.toRelease = toRelease;
@@ -515,7 +546,7 @@ public abstract class AbstractLookupService<R extends AbstractLookupService.Requ
 
         @Override
         public final String[] indices() {
-            return new String[] { shardId.getIndexName() };
+            return new String[] { indexPattern };
         }
 
         @Override

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

@@ -11,6 +11,7 @@ import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionListenerResponseHandler;
 import org.elasticsearch.action.support.ContextPreservingActionListener;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -30,6 +31,8 @@ import org.elasticsearch.index.mapper.RangeFieldMapper;
 import org.elasticsearch.index.mapper.RangeType;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.search.internal.AliasFilter;
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.transport.TransportRequestOptions;
@@ -66,16 +69,20 @@ public class EnrichLookupService extends AbstractLookupService<EnrichLookupServi
 
     public EnrichLookupService(
         ClusterService clusterService,
+        IndicesService indicesService,
         LookupShardContextFactory lookupShardContextFactory,
         TransportService transportService,
+        IndexNameExpressionResolver indexNameExpressionResolver,
         BigArrays bigArrays,
         BlockFactory blockFactory
     ) {
         super(
             LOOKUP_ACTION_NAME,
             clusterService,
+            indicesService,
             lookupShardContextFactory,
             transportService,
+            indexNameExpressionResolver,
             bigArrays,
             blockFactory,
             true,
@@ -102,14 +109,15 @@ public class EnrichLookupService extends AbstractLookupService<EnrichLookupServi
     protected QueryList queryList(
         TransportRequest request,
         SearchExecutionContext context,
+        AliasFilter aliasFilter,
         Block inputBlock,
         @Nullable DataType inputDataType
     ) {
         MappedFieldType fieldType = context.getFieldType(request.matchField);
         validateTypes(inputDataType, fieldType);
         return switch (request.matchType) {
-            case "match", "range" -> termQueryList(fieldType, context, inputBlock, inputDataType);
-            case "geo_match" -> QueryList.geoShapeQueryList(fieldType, context, inputBlock);
+            case "match", "range" -> termQueryList(fieldType, context, aliasFilter, inputBlock, inputDataType);
+            case "geo_match" -> QueryList.geoShapeQueryList(fieldType, context, aliasFilter, inputBlock);
             default -> throw new EsqlIllegalArgumentException("illegal match type " + request.matchType);
         };
     }
@@ -168,7 +176,7 @@ public class EnrichLookupService extends AbstractLookupService<EnrichLookupServi
             List<NamedExpression> extractFields,
             Source source
         ) {
-            super(sessionId, index, inputDataType, inputPage, extractFields, source);
+            super(sessionId, index, index, inputDataType, inputPage, extractFields, source);
             this.matchType = matchType;
             this.matchField = matchField;
         }
@@ -189,7 +197,7 @@ public class EnrichLookupService extends AbstractLookupService<EnrichLookupServi
             List<NamedExpression> extractFields,
             Source source
         ) {
-            super(sessionId, shardId, inputDataType, inputPage, toRelease, extractFields, source);
+            super(sessionId, shardId, shardId.getIndexName(), inputDataType, inputPage, toRelease, extractFields, source);
             this.matchType = matchType;
             this.matchField = matchField;
         }

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

@@ -43,6 +43,7 @@ public final class LookupFromIndexOperator extends AsyncOperator<LookupFromIndex
         int inputChannel,
         Function<DriverContext, LookupFromIndexService> lookupService,
         DataType inputDataType,
+        String lookupIndexPattern,
         String lookupIndex,
         String matchField,
         List<NamedExpression> loadFields,
@@ -73,6 +74,7 @@ public final class LookupFromIndexOperator extends AsyncOperator<LookupFromIndex
                 inputChannel,
                 lookupService.apply(driverContext),
                 inputDataType,
+                lookupIndexPattern,
                 lookupIndex,
                 matchField,
                 loadFields,
@@ -86,6 +88,7 @@ public final class LookupFromIndexOperator extends AsyncOperator<LookupFromIndex
     private final CancellableTask parentTask;
     private final int inputChannel;
     private final DataType inputDataType;
+    private final String lookupIndexPattern;
     private final String lookupIndex;
     private final String matchField;
     private final List<NamedExpression> loadFields;
@@ -108,6 +111,7 @@ public final class LookupFromIndexOperator extends AsyncOperator<LookupFromIndex
         int inputChannel,
         LookupFromIndexService lookupService,
         DataType inputDataType,
+        String lookupIndexPattern,
         String lookupIndex,
         String matchField,
         List<NamedExpression> loadFields,
@@ -119,6 +123,7 @@ public final class LookupFromIndexOperator extends AsyncOperator<LookupFromIndex
         this.inputChannel = inputChannel;
         this.lookupService = lookupService;
         this.inputDataType = inputDataType;
+        this.lookupIndexPattern = lookupIndexPattern;
         this.lookupIndex = lookupIndex;
         this.matchField = matchField;
         this.loadFields = loadFields;
@@ -132,6 +137,7 @@ public final class LookupFromIndexOperator extends AsyncOperator<LookupFromIndex
         LookupFromIndexService.Request request = new LookupFromIndexService.Request(
             sessionId,
             lookupIndex,
+            lookupIndexPattern,
             inputDataType,
             matchField,
             new Page(inputBlock),

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

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.enrich;
 
 import org.elasticsearch.TransportVersions;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.collect.Iterators;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -22,8 +23,11 @@ import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.search.internal.AliasFilter;
 import org.elasticsearch.tasks.TaskId;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.esql.EsqlIllegalArgumentException;
 import org.elasticsearch.xpack.esql.action.EsqlQueryAction;
 import org.elasticsearch.xpack.esql.core.expression.NamedExpression;
 import org.elasticsearch.xpack.esql.core.tree.Source;
@@ -45,16 +49,20 @@ public class LookupFromIndexService extends AbstractLookupService<LookupFromInde
 
     public LookupFromIndexService(
         ClusterService clusterService,
+        IndicesService indicesService,
         LookupShardContextFactory lookupShardContextFactory,
         TransportService transportService,
+        IndexNameExpressionResolver indexNameExpressionResolver,
         BigArrays bigArrays,
         BlockFactory blockFactory
     ) {
         super(
             LOOKUP_ACTION_NAME,
             clusterService,
+            indicesService,
             lookupShardContextFactory,
             transportService,
+            indexNameExpressionResolver,
             bigArrays,
             blockFactory,
             false,
@@ -67,6 +75,7 @@ public class LookupFromIndexService extends AbstractLookupService<LookupFromInde
         return new TransportRequest(
             request.sessionId,
             shardId,
+            request.indexPattern,
             request.inputDataType,
             request.inputPage,
             null,
@@ -80,10 +89,11 @@ public class LookupFromIndexService extends AbstractLookupService<LookupFromInde
     protected QueryList queryList(
         TransportRequest request,
         SearchExecutionContext context,
+        AliasFilter aliasFilter,
         Block inputBlock,
         @Nullable DataType inputDataType
     ) {
-        return termQueryList(context.getFieldType(request.matchField), context, inputBlock, inputDataType).onlySingleValues();
+        return termQueryList(context.getFieldType(request.matchField), context, aliasFilter, inputBlock, inputDataType).onlySingleValues();
     }
 
     @Override
@@ -102,13 +112,14 @@ public class LookupFromIndexService extends AbstractLookupService<LookupFromInde
         Request(
             String sessionId,
             String index,
+            String indexPattern,
             DataType inputDataType,
             String matchField,
             Page inputPage,
             List<NamedExpression> extractFields,
             Source source
         ) {
-            super(sessionId, index, inputDataType, inputPage, extractFields, source);
+            super(sessionId, index, indexPattern, inputDataType, inputPage, extractFields, source);
             this.matchField = matchField;
         }
     }
@@ -119,6 +130,7 @@ public class LookupFromIndexService extends AbstractLookupService<LookupFromInde
         TransportRequest(
             String sessionId,
             ShardId shardId,
+            String indexPattern,
             DataType inputDataType,
             Page inputPage,
             Page toRelease,
@@ -126,7 +138,7 @@ public class LookupFromIndexService extends AbstractLookupService<LookupFromInde
             String matchField,
             Source source
         ) {
-            super(sessionId, shardId, inputDataType, inputPage, toRelease, extractFields, source);
+            super(sessionId, shardId, indexPattern, inputDataType, inputPage, toRelease, extractFields, source);
             this.matchField = matchField;
         }
 
@@ -134,6 +146,14 @@ public class LookupFromIndexService extends AbstractLookupService<LookupFromInde
             TaskId parentTaskId = TaskId.readFromStream(in);
             String sessionId = in.readString();
             ShardId shardId = new ShardId(in);
+
+            String indexPattern;
+            if (in.getTransportVersion().onOrAfter(TransportVersions.JOIN_ON_ALIASES_8_19)) {
+                indexPattern = in.readString();
+            } else {
+                indexPattern = shardId.getIndexName();
+            }
+
             DataType inputDataType = DataType.fromTypeName(in.readString());
             Page inputPage;
             try (BlockStreamInput bsi = new BlockStreamInput(in, blockFactory)) {
@@ -149,6 +169,7 @@ public class LookupFromIndexService extends AbstractLookupService<LookupFromInde
             TransportRequest result = new TransportRequest(
                 sessionId,
                 shardId,
+                indexPattern,
                 inputDataType,
                 inputPage,
                 inputPage,
@@ -165,6 +186,13 @@ public class LookupFromIndexService extends AbstractLookupService<LookupFromInde
             super.writeTo(out);
             out.writeString(sessionId);
             out.writeWriteable(shardId);
+
+            if (out.getTransportVersion().onOrAfter(TransportVersions.JOIN_ON_ALIASES_8_19)) {
+                out.writeString(indexPattern);
+            } else if (indexPattern.equals(shardId.getIndexName()) == false) {
+                throw new EsqlIllegalArgumentException("Aliases and index patterns are not allowed for LOOKUP JOIN []", indexPattern);
+            }
+
             out.writeString(inputDataType.typeName());
             out.writeWriteable(inputPage);
             PlanStreamOutput planOut = new PlanStreamOutput(out, null);

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plan/logical/join/LookupJoin.java

@@ -102,11 +102,6 @@ public class LookupJoin extends Join implements SurrogateLogicalPlan, PostAnalys
                     )
                 );
             }
-
-            // this check is crucial for security: ES|QL would use the concrete indices, so it would bypass the security on the alias
-            if (esr.concreteIndices().contains(esr.indexPattern()) == false) {
-                failures.add(fail(this, "Aliases and index patterns are not allowed for LOOKUP JOIN [{}]", esr.indexPattern()));
-            }
         });
     }
 }

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/planner/LocalExecutionPlanner.java

@@ -614,6 +614,7 @@ public class LocalExecutionPlanner {
                 matchConfig.channel(),
                 ctx -> lookupFromIndexService,
                 matchConfig.type(),
+                localSourceExec.indexPattern(),
                 indexName,
                 matchConfig.fieldName(),
                 join.addedFields().stream().map(f -> (NamedExpression) f).toList(),

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

@@ -114,15 +114,19 @@ public class TransportEsqlQueryAction extends HandledTransportAction<EsqlQueryRe
             .fromSearchService(searchService);
         this.enrichLookupService = new EnrichLookupService(
             clusterService,
+            searchService.getIndicesService(),
             lookupLookupShardContextFactory,
             transportService,
+            indexNameExpressionResolver,
             bigArrays,
             blockFactoryProvider.blockFactory()
         );
         this.lookupFromIndexService = new LookupFromIndexService(
             clusterService,
+            searchService.getIndicesService(),
             lookupLookupShardContextFactory,
             transportService,
+            indexNameExpressionResolver,
             bigArrays,
             blockFactoryProvider.blockFactory()
         );

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

@@ -15,6 +15,7 @@ import org.apache.lucene.store.Directory;
 import org.apache.lucene.tests.index.RandomIndexWriter;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.action.support.replication.ClusterStateCreationUtils;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
 import org.elasticsearch.cluster.service.ClusterService;
@@ -44,6 +45,8 @@ import org.elasticsearch.index.mapper.MapperService;
 import org.elasticsearch.index.mapper.MapperServiceTestCase;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.indices.CrankyCircuitBreakerService;
+import org.elasticsearch.indices.IndicesService;
+import org.elasticsearch.indices.TestIndexNameExpressionResolver;
 import org.elasticsearch.search.internal.AliasFilter;
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.TaskId;
@@ -74,6 +77,7 @@ import java.util.stream.LongStream;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.matchesPattern;
+import static org.mockito.Mockito.mock;
 
 public class LookupFromIndexOperatorTests extends OperatorTestCase {
     private static final int LOOKUP_SIZE = 1000;
@@ -144,6 +148,7 @@ public class LookupFromIndexOperatorTests extends OperatorTestCase {
             this::lookupService,
             inputDataType,
             lookupIndex,
+            lookupIndex,
             matchField,
             loadFields,
             Source.EMPTY
@@ -175,6 +180,8 @@ public class LookupFromIndexOperatorTests extends OperatorTestCase {
                 .build(),
             ClusterSettings.createBuiltInClusterSettings()
         );
+        IndicesService indicesService = mock(IndicesService.class);
+        IndexNameExpressionResolver indexNameExpressionResolver = TestIndexNameExpressionResolver.newInstance();
         releasables.add(clusterService::stop);
         ClusterServiceUtils.setState(clusterService, ClusterStateCreationUtils.state("idx", 1, 1));
         if (beCranky) {
@@ -185,8 +192,10 @@ public class LookupFromIndexOperatorTests extends OperatorTestCase {
         BlockFactory blockFactory = ctx.blockFactory();
         return new LookupFromIndexService(
             clusterService,
+            indicesService,
             lookupShardContextFactory(),
             transportService(clusterService),
+            indexNameExpressionResolver,
             bigArrays,
             blockFactory
         );

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/190_lookup_join.yml

@@ -6,7 +6,7 @@ setup:
         - method: POST
           path: /_query
           parameters: []
-          capabilities: [join_lookup_v12]
+          capabilities: [join_lookup_v12, enable_lookup_join_on_aliases]
       reason: "uses LOOKUP JOIN"
   - do:
       indices.create:
@@ -120,8 +120,6 @@ non-lookup index:
 
 ---
 alias-repeated-alias:
-  - skip:
-      awaits_fix: "LOOKUP JOIN does not support index aliases for now"
   - do:
       esql.query:
         body:
@@ -136,8 +134,6 @@ alias-repeated-alias:
 
 ---
 alias-repeated-index:
-  - skip:
-      awaits_fix: "LOOKUP JOIN does not support index aliases for now"
   - do:
       esql.query:
         body:
@@ -152,8 +148,6 @@ alias-repeated-index:
 
 ---
 alias-pattern-multiple:
-  - skip:
-      awaits_fix: "LOOKUP JOIN does not support index aliases for now"
   - do:
       esql.query:
         body:
@@ -165,8 +159,6 @@ alias-pattern-multiple:
 
 ---
 alias-pattern-single:
-  - skip:
-      awaits_fix: "LOOKUP JOIN does not support index aliases for now"
   - do:
       esql.query:
         body:

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/esql/191_lookup_join_on_datastreams.yml

@@ -6,7 +6,7 @@ setup:
         - method: POST
           path: /_query
           parameters: []
-          capabilities: [lookup_join_no_aliases]
+          capabilities: [enable_lookup_join_on_aliases]
       reason: "uses LOOKUP JOIN"
 
   - do:
@@ -30,6 +30,8 @@ setup:
                    type: date
                 x:
                   type: keyword
+                y:
+                  type: keyword
 
   - do:
       indices.put_index_template:
@@ -40,29 +42,33 @@ setup:
           composed_of: [ "my_mappings", "my_settings" ]
           priority: 500
 
+  - do:
+      indices.create_data_stream:
+        name: my_data_stream
 
   - do:
-      bulk:
-        index: "my_data_stream"
-        refresh: true
+      index:
+        index:  my_data_stream
         body:
-          - { "index": { } }
-          - { "x": "foo", "y": "y1" }
-          - { "index": { } }
-          - { "x": "bar", "y": "y2" }
-
-
+          '@timestamp': '2020-12-12'
+          'x': 'foo'
+          'y': 'y1'
 
+  - do:
+      indices.refresh:
+        index: my_data_stream
 ---
-"data streams not supported in LOOKUP JOIN":
+"data streams supported in LOOKUP JOIN":
   - do:
       esql.query:
         body:
-          query: 'row x = "foo" | LOOKUP JOIN my_data_stream ON x'
-      catch: "bad_request"
+          query: 'ROW x = "foo" | LOOKUP JOIN my_data_stream ON x | KEEP x, y | LIMIT 1'
 
-  - match: { error.type: "verification_exception" }
-  - contains: { error.reason: "Found 1 problem\nline 1:17: Aliases and index patterns are not allowed for LOOKUP JOIN [my_data_stream]" }
+  - match: {columns.0.name: "x"}
+  - match: {columns.0.type: "keyword"}
+  - match: {columns.1.name: "y"}
+  - match: {columns.1.type: "keyword"}
+  - match: {values.0: ["foo", "y1"]}