[ES|QL] Assign new id to alias created by ReplaceMissingFieldWithNull when there is lookup join

docs/changelog/125462.yaml

@@ -0,0 +1,6 @@
+pr: 125462
+summary: Assign new id to alias created by `ReplaceMissingFieldWithNull` when there
+  is lookup join
+area: ES|QL
+type: bug
+issues: []

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

@@ -986,6 +986,32 @@ public void testDoubleParamsWithLookupJoin() throws IOException {
         );
     }
 
+    public void testMultipleBatchesWithLookupJoin() throws IOException {
+        assumeTrue(
+            "Requires new null alias ids for join with multiple batches",
+            EsqlCapabilities.Cap.REPLACE_MISSING_FIELD_WITH_NULL_NEW_ALIAS_ID_FOR_JOIN_AND_MULTIPLE_BATCHES.isEnabled()
+        );
+        // Create more than 10 indices to trigger multiple batches of data node execution.
+        // The sort field should be missing on some indices to reproduce NullPointerException caused by duplicated items in layout
+        for (int i = 1; i <= 20; i++) {
+            createIndex("idx" + i, randomBoolean(), "\"mappings\": {\"properties\" : {\"a\" : {\"type\" : \"keyword\"}}}");
+        }
+        bulkLoadTestDataLookupMode(10);
+        // lookup join with and without sort
+        for (String sort : List.of("", "| sort integer")) {
+            var query = requestObjectBuilder().query(format(null, "from * | lookup join {} on integer {}", testIndexName(), sort));
+            Map<String, Object> result = runEsql(query);
+            var columns = as(result.get("columns"), List.class);
+            assertEquals(21, columns.size());
+            var values = as(result.get("values"), List.class);
+            assertEquals(10, values.size());
+        }
+        // clean up
+        for (int i = 1; i <= 20; i++) {
+            assertThat(deleteIndex("idx" + i).isAcknowledged(), is(true));
+        }
+    }
+
     private void validateResultsOfDoubleParametersForIdentifiers(RequestObjectBuilder query) throws IOException {
         Map<String, Object> result = runEsql(query);
         Map<String, String> colA = Map.of("name", "boolean", "type", "boolean");
@@ -1668,6 +1694,13 @@ private static String repeatValueAsMV(Object value) {
         return "[" + value + ", " + value + "]";
     }
 
+    private static void createIndex(String indexName, boolean lookupMode, String mapping) throws IOException {
+        Request request = new Request("PUT", "/" + indexName);
+        String settings = "\"settings\" : {\"mode\" : \"lookup\"}, ";
+        request.setJsonEntity("{" + (lookupMode ? settings : "") + mapping + "}");
+        assertEquals(200, client().performRequest(request).getStatusLine().getStatusCode());
+    }
+
     public static RequestObjectBuilder requestObjectBuilder() throws IOException {
         return new RequestObjectBuilder();
     }

x-pack/plugin/esql/qa/testFixtures/src/main/resources/lookup-join.csv-spec

@@ -1452,3 +1452,20 @@ emp_no:integer | language_code:integer | language_name:keyword
 10092          | 1                     | English
 10093          | 3                     | Spanish
 ;
+
+multipleBatches
+required_capability: join_lookup_v12
+required_capability: remove_redundant_sort
+required_capability: replace_missing_field_with_null_new_alias_id_for_join_and_multiple_batches
+
+from *
+| rename city.country.continent.planet.name as message
+| lookup join message_types_lookup on message
+| sort language_code, birth_date
+| keep birth_date
+| limit 1
+;
+
+birth_date:date
+null
+;

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

@@ -918,7 +918,12 @@ public enum Cap {
         /**
          * Index component selector syntax (my-data-stream-name::failures)
          */
-        INDEX_COMPONENT_SELECTORS(DataStream.isFailureStoreFeatureFlagEnabled());
+        INDEX_COMPONENT_SELECTORS(DataStream.isFailureStoreFeatureFlagEnabled()),
+
+        /**
+         * Create null alias with new id in ReplaceMissingFieldWithNull when there is lookup join with multiple batches.
+         */
+        REPLACE_MISSING_FIELD_WITH_NULL_NEW_ALIAS_ID_FOR_JOIN_AND_MULTIPLE_BATCHES;
 
         private final boolean enabled;
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/optimizer/rules/logical/local/ReplaceMissingFieldWithNull.java

@@ -81,7 +81,11 @@ else if (plan instanceof Project project) {
                     Alias nullAlias = nullLiteral.get(f.dataType());
                     // save the first field as null (per datatype)
                     if (nullAlias == null) {
-                        Alias alias = new Alias(f.source(), f.name(), Literal.of(f, null), f.id());
+                        // In case of batch executions on data nodes and join exists, SearchStats may not always be available for all,
+                        // fields, creating a new alias for null with the same id as the field id can potentially cause planEval to add a
+                        // duplicated ChannelSet to a layout, and Layout.builder().build() could throw a NullPointerException.
+                        // As a workaround, assign a new alias id to the null alias when join exists and SearchStats is not available.
+                        Alias alias = new Alias(f.source(), f.name(), Literal.of(f, null), joinAttributes.isEmpty() ? f.id() : null);
                         nullLiteral.put(dt, alias);
                         projection = alias.toAttribute();
                     }