assign new id to aliases created by ReplaceMissingFieldWithNull

Author: fang-xing-esql

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

@@ -986,6 +986,24 @@ public void testDoubleParamsWithLookupJoin() throws IOException {
         );
     }
 
+    public void testMultipleBatchesWithLookupJoin() throws IOException {
+        // create 20 indices to trigger multiple batches of data node planning and execution
+        for (int i = 1; i <= 20; i++) {
+            createIndex("idx" + i, false);
+        }
+        bulkLoadTestDataLookupMode(10);
+        var query = requestObjectBuilder().query(format(null, "from * | lookup join {} on integer | sort integer", testIndexName()));
+        Map<String, Object> result = runEsql(query);
+        var columns = as(result.get("columns"), List.class);
+        assertEquals(20, 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");

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

@@ -1452,3 +1452,29 @@ 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
+
+from *
+| dissect street "%{height_range} %{MNyXV}"
+| rename env AS kxpCK, pod AS etUHW, language_code AS city.country.continent.planet.galaxy
+| rename city.country.continent.planet.name as message
+| lookup join message_types_lookup on message
+| rename languages.int as language_code
+| lookup join languages_lookup on language_code
+| rename language_code as language_code
+| lookup join languages_lookup on language_code
+| stats c = count(*) by language_code
+| sort language_code
+;
+
+c:long | language_code:long
+15     | 1
+19     | 2
+17     | 3
+18     | 4
+21     | 5
+6571   | null
+;

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

@@ -81,7 +81,9 @@ 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());
+                        Alias alias = joinAttributes.isEmpty()
+                            ? new Alias(f.source(), f.name(), Literal.of(f, null), f.id())
+                            : new Alias(f.source(), f.name(), Literal.of(f, null));
                         nullLiteral.put(dt, alias);
                         projection = alias.toAttribute();
                     }