more tests and comments

Author: fang-xing-esql

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

@@ -992,12 +992,15 @@ public void testMultipleBatchesWithLookupJoin() throws IOException {
             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());
+        // 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(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));

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

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